Send the rest at once after the first passes, and resume a stored gate (hq ADR 0353)
mesh/merge-gate head c2ea9b4c5187 merged into trunk 1601cb17432e — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head c2ea9b4c5187 merged into trunk 1601cb17432e — fail: its merge-check.sh failed: --- FAIL: TestAReleaseSendsTheRestAtOnceAfterTheFirstPa…
mesh/delivery superseded: a newer head of the same pull request
mesh/merge-gate head c2ea9b4c5187 merged into trunk 1601cb17432e — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head c2ea9b4c5187 merged into trunk 1601cb17432e — fail: its merge-check.sh failed: --- FAIL: TestAReleaseSendsTheRestAtOnceAfterTheFirstPa…
mesh/delivery superseded: a newer head of the same pull request
A release now sends every other machine in one step once its first machine passed, each judged by its own gate; a failing one is put back without stopping the others, a silent one is left. A send is kept before it is made, and a machine already sent what a gate is for is not sent again: the gate counts from that send and takes over another walk's readings of the same build there.
This commit is contained in:
@@ -1532,6 +1532,12 @@ func gateLine(g *inventory.PlanGate) string {
|
||||
case inventory.GateFailed:
|
||||
what = "FAILED"
|
||||
}
|
||||
switch {
|
||||
case g.Verdict == "" && g.Pending:
|
||||
what = "kept, not yet sent" // novox/hq ADR 0353 §8
|
||||
case g.Verdict == "" && g.Silent:
|
||||
what = "silent on its send, left" // ADR 0353 §4
|
||||
}
|
||||
if g.Verdict == "" && g.Waiting() {
|
||||
what = "waits for a busy container on " + g.Busy.Machine
|
||||
}
|
||||
|
||||
@@ -151,7 +151,7 @@ func heldMachines(ctx context.Context, open *stores, names []string) (map[string
|
||||
for _, mv := range f.moves(node, modules, same, known, false) {
|
||||
if planStillToSend(plans, mv.Module, node) == "" {
|
||||
why = append(why, fmt.Sprintf("%s would move from %s to %s, which has passed no gate yet — a release "+
|
||||
"plan sends it, one machine at a time, judged", mv.Module, buildName(mv.From), buildName(mv.To)))
|
||||
"plan sends it, one machine first and then the rest at once, judged", mv.Module, buildName(mv.From), buildName(mv.To)))
|
||||
}
|
||||
}
|
||||
if len(why) > 0 {
|
||||
|
||||
+424
-41
@@ -39,8 +39,8 @@ import (
|
||||
// from the same manifest is no move.
|
||||
//
|
||||
// **The release plan** is what walks the waiting moves through: whenever moves wait for a gate that no
|
||||
// started plan is walking, the mesh opens one — every such machine, one at a time, the control node last,
|
||||
// each sent and judged before the next. A release plan that fails stops the next one opening on its own:
|
||||
// started plan is walking, the mesh opens one — every such machine, the first sent and judged, then the rest
|
||||
// at once, each judged by its own gate (novox/hq ADR 0353, in place of one at a time). A release plan that fails stops the next one opening on its own:
|
||||
// a person releases it again (`upgrade release-backlog --why`), after the condition says why.
|
||||
|
||||
// sendScope is what a send may carry that no gate has seen.
|
||||
@@ -197,13 +197,19 @@ func (f moveFacts) walkedBy(module, node, to string) string {
|
||||
continue
|
||||
}
|
||||
if p.Release != nil {
|
||||
g := p.Release.Gate
|
||||
if g == nil || g.Verdict != "" {
|
||||
continue
|
||||
// The first machine's gate, and every gate of the rest still judging (novox/hq ADR 0353).
|
||||
gates := []*inventory.PlanGate{p.Release.Gate}
|
||||
for _, g := range p.Release.Gates {
|
||||
gates = append(gates, g)
|
||||
}
|
||||
for _, c := range g.Carried {
|
||||
if c.Module == module && !(slices.Contains(g.Machines, node) && sameCommit(c.To, to)) {
|
||||
return p.ID
|
||||
for _, g := range gates {
|
||||
if g == nil || g.Verdict != "" || g.Silent {
|
||||
continue
|
||||
}
|
||||
for _, c := range g.Carried {
|
||||
if c.Module == module && !(slices.Contains(g.Machines, node) && sameCommit(c.To, to)) {
|
||||
return p.ID
|
||||
}
|
||||
}
|
||||
}
|
||||
continue
|
||||
@@ -303,8 +309,8 @@ func ungatedIn(ctx context.Context, open *stores, names []string, addedHolder st
|
||||
}
|
||||
}
|
||||
if len(refused) > 0 {
|
||||
return nil, fmt.Errorf("%w: %s — a walk sends them, one machine at a time, each judged (novox/hq ADR "+
|
||||
"0236); `upgrade backlog` lists them", errUngated, strings.Join(refused, "; "))
|
||||
return nil, fmt.Errorf("%w: %s — a walk sends them, one machine first and then the rest at once, each judged (novox/hq ADR "+
|
||||
"0236, ADR 0353); `upgrade backlog` lists them", errUngated, strings.Join(refused, "; "))
|
||||
}
|
||||
return kept, nil
|
||||
}
|
||||
@@ -320,19 +326,35 @@ func ungatedIn(ctx context.Context, open *stores, names []string, addedHolder st
|
||||
// And the grants step first, when the send changes what the bus's user list must say (novox/hq issue 490):
|
||||
// answered for the gate to keep, nil when there was none.
|
||||
func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.CarriedMove) ([]inventory.CarriedMove, []string, *inventory.GrantsStep, error) {
|
||||
inv := open.inventory
|
||||
f, err := readMoveFacts(ctx, inv)
|
||||
moves, err := gatedMoves(ctx, open, node, owns)
|
||||
if err != nil || len(moves) == 0 {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
sayRecreations(ctx, open, moves)
|
||||
sent, grants, err := sendJudged(ctx, open, node)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
return moves, sent, grants, nil
|
||||
}
|
||||
|
||||
// gatedMoves is what a gated send to one machine would carry, with the build each moves to, made of nothing:
|
||||
// gatedSend's reading, which a walk also keeps before it sends (novox/hq ADR 0353 §8). Refused the same way
|
||||
// while a plan that has started walks one of those builds elsewhere.
|
||||
func gatedMoves(ctx context.Context, open *stores, node string, owns []inventory.CarriedMove) ([]inventory.CarriedMove, error) {
|
||||
inv := open.inventory
|
||||
f, err := readMoveFacts(ctx, inv)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
moves, err := machineMoves(ctx, open, f, node, true)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
// A move the send keeps is not made, and is no move for the gate to judge (novox/hq issue 201): a core module
|
||||
// kept over an older record judged as moving back to it would read the gate as superseded by its own send.
|
||||
if moves, err = withoutKept(ctx, open, node, moves); err != nil {
|
||||
return nil, nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
own := func(module string) bool {
|
||||
return slices.ContainsFunc(owns, func(o inventory.CarriedMove) bool { return o.Module == module })
|
||||
@@ -342,7 +364,7 @@ func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.
|
||||
continue
|
||||
}
|
||||
if id := f.walkedBy(mv.Module, node, mv.To); id != "" {
|
||||
return nil, nil, nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
|
||||
return nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
|
||||
mv.Module, short(mv.To), node, id)
|
||||
}
|
||||
}
|
||||
@@ -350,7 +372,7 @@ func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.
|
||||
// plan, or a release, on this machine is not crossed with a newer build; this send waits for its gate.
|
||||
for _, o := range owns {
|
||||
if id := f.walkedBy(o.Module, node, o.To); id != "" {
|
||||
return nil, nil, nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
|
||||
return nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
|
||||
o.Module, short(o.To), node, id)
|
||||
}
|
||||
}
|
||||
@@ -363,30 +385,26 @@ func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.
|
||||
moves[i].Build = o.Build
|
||||
}
|
||||
}
|
||||
if len(moves) == 0 && len(owns) == 0 {
|
||||
return nil, nil, nil, nil
|
||||
if len(moves) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
for i := range moves {
|
||||
if moves[i].Build == "" {
|
||||
if moves[i].Build, err = inv.BuildOf(ctx, moves[i].Module, moves[i].To); err != nil {
|
||||
return nil, nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
sort.Slice(moves, func(i, j int) bool { return moves[i].Module < moves[j].Module })
|
||||
sayRecreations(ctx, open, moves)
|
||||
sent, grants, err := sendJudged(ctx, open, node)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
return moves, sent, grants, nil
|
||||
return moves, nil
|
||||
}
|
||||
|
||||
// passCarried keeps a pass as the verdict of every build the gate judged beside its own module.
|
||||
func passCarried(ctx context.Context, open *stores, p *inventory.Plan, g *inventory.PlanGate, except string) {
|
||||
seen := map[string]bool{}
|
||||
for _, c := range g.Carried {
|
||||
if c.Module == except || c.Build == "" || seen[c.Build] {
|
||||
// A build put back with another machine's failure is not this gate's to pass (novox/hq ADR 0353 §3).
|
||||
if c.Module == except || c.Build == "" || seen[c.Build] || slices.Contains(g.Returned, c.Module) {
|
||||
continue
|
||||
}
|
||||
seen[c.Build] = true
|
||||
@@ -695,7 +713,7 @@ func releaseBacklog(ctx context.Context, open *stores, by string) (*inventory.Pl
|
||||
p := inventory.Plan{ID: fmt.Sprintf("release-%d", now.UnixNano()), Repository: releaseRepository,
|
||||
Created: now, State: inventory.PlanRolling, Tiers: [][]string{names}, Modules: map[string]*inventory.PlanModule{},
|
||||
Release: &inventory.PlanRelease{Order: order, By: by},
|
||||
Note: fmt.Sprintf("%d build(s) wait for a gate on %s; one machine at a time, each judged", len(names),
|
||||
Note: fmt.Sprintf("%d build(s) wait for a gate on %s; one machine first, then the rest at once, each judged", len(names),
|
||||
strings.Join(order, ", "))}
|
||||
if err := inv.SavePlan(ctx, &p); err != nil {
|
||||
return nil, err
|
||||
@@ -704,14 +722,22 @@ func releaseBacklog(ctx context.Context, open *stores, by string) (*inventory.Pl
|
||||
return &p, nil
|
||||
}
|
||||
|
||||
// advanceRelease takes one step of a release plan: the next machine sent everything waiting there under a
|
||||
// gate, or the machine being judged judged once more; the plan done when every machine is.
|
||||
// advanceRelease takes one step of a release plan: the first machine sent everything waiting there under a
|
||||
// gate and judged; once it passed, every other machine sent at once, each under its own gate (novox/hq ADR
|
||||
// 0353 §1), and judged; the plan done when every machine is.
|
||||
//
|
||||
// **A send is kept before it is made** (ADR 0353 §8): the step that chooses a machine keeps its gate pending with
|
||||
// what it will carry, and the next step makes the send. A controller replaced between the send and keeping it
|
||||
// finds the gate pending, reads the machine's last send, and neither sends again nor leaves the machine unjudged.
|
||||
func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool, error) {
|
||||
r := p.Release
|
||||
if r.Settings != nil {
|
||||
return advanceSettingsWalk(ctx, open, p)
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
if r.Rest {
|
||||
return advanceRest(ctx, open, p, now)
|
||||
}
|
||||
if r.Gate == nil {
|
||||
heard, err := releaseHeard(ctx, open)
|
||||
if err != nil {
|
||||
@@ -724,7 +750,7 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
|
||||
r.Next++
|
||||
continue
|
||||
}
|
||||
moves, sent, grants, err := gatedSend(ctx, open, node, nil)
|
||||
moves, err := gatedMoves(ctx, open, node, nil)
|
||||
if errors.Is(err, errWalkedElsewhere) {
|
||||
note := fmt.Sprintf("waiting before %s: %v", node, err)
|
||||
changed := p.Note != note
|
||||
@@ -732,23 +758,15 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
|
||||
return changed, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("sending %s what waits there: %w", node, err)
|
||||
return false, fmt.Errorf("reading what waits on %s: %w", node, err)
|
||||
}
|
||||
if len(moves) == 0 {
|
||||
r.Done = append(r.Done, node)
|
||||
r.Next++
|
||||
continue
|
||||
}
|
||||
r.Gate = &inventory.PlanGate{Machines: sent, Since: &now, Carried: moves, Sent: sentNow(ctx, open.inventory, sent),
|
||||
Grants: grants}
|
||||
p.Note = fmt.Sprintf("sent %s %d build(s) that waited for a gate; judging them there", node, len(moves))
|
||||
if said := grantsSaid(grants); said != "" {
|
||||
p.Note += "; " + said
|
||||
}
|
||||
if said := recreationsSaid(moves); said != "" {
|
||||
p.Note += "; " + said
|
||||
}
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
r.Gate = &inventory.PlanGate{Machines: []string{node}, Carried: moves, Pending: true}
|
||||
p.Note = fmt.Sprintf("sending %s %d build(s) that wait for a gate, first", node, len(moves))
|
||||
return true, nil
|
||||
}
|
||||
p.State, p.Tier = inventory.PlanDone, len(p.Tiers)
|
||||
@@ -759,6 +777,19 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
|
||||
return true, nil
|
||||
}
|
||||
g := r.Gate
|
||||
if g.Pending {
|
||||
node := firstOf(g.Machines)
|
||||
out, err := sendKept(ctx, open, p, node, g, now)
|
||||
if err != nil || out == keptWaiting {
|
||||
return out == keptWaiting, err
|
||||
}
|
||||
if out == keptNothing {
|
||||
r.Gate = nil
|
||||
r.Done = append(r.Done, node)
|
||||
r.Next++
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
verdict, err := judgeMoves(ctx, open, g, nil, now)
|
||||
if err != nil {
|
||||
return false, err
|
||||
@@ -777,6 +808,11 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
|
||||
r.Done = append(r.Done, firstOf(g.Machines))
|
||||
r.Next++
|
||||
r.Gate = nil
|
||||
// **Then the rest at once** (novox/hq ADR 0353 §1): every other machine heard from, each judged by its own
|
||||
// gate, in place of one machine at a time (ADR 0236 §22).
|
||||
r.Rest = true
|
||||
p.Note = fmt.Sprintf("passed on %s; the rest are sent at once, each judged", firstOf(g.Machines))
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
return true, nil
|
||||
case inventory.GateSuperseded:
|
||||
// Another send moved a judged module on the judged machine (novox/hq issue 352): no verdict on what
|
||||
@@ -797,6 +833,353 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// What sendKept did with a kept send.
|
||||
const (
|
||||
keptSent = iota // sent now
|
||||
keptAlready // the machine was already sent it: judged from that send, nothing sent
|
||||
keptNothing // nothing waits there any more: no gate
|
||||
keptWaiting // another walk judges a build it would carry: it waits
|
||||
)
|
||||
|
||||
// sendKept makes a send a gate kept before it was made (novox/hq ADR 0353 §8) — unless the machine's last send
|
||||
// already carries what the gate is for, made before a controller was replaced and before it was kept: then
|
||||
// nothing is sent, and the gate judges from that send (§6).
|
||||
func sendKept(ctx context.Context, open *stores, p *inventory.Plan, node string, g *inventory.PlanGate,
|
||||
now time.Time) (int, error) {
|
||||
inv := open.inventory
|
||||
moves, sent, grants, err := gatedSend(ctx, open, node, nil)
|
||||
if errors.Is(err, errWalkedElsewhere) {
|
||||
p.Note = fmt.Sprintf("waiting before %s: %v", node, err)
|
||||
return keptWaiting, nil
|
||||
}
|
||||
if err != nil {
|
||||
return keptSent, fmt.Errorf("sending %s what waits there: %w", node, err)
|
||||
}
|
||||
if len(moves) > 0 {
|
||||
g.Machines, g.Carried, g.Since, g.Grants, g.Pending = sent, moves, &now, grants, false
|
||||
g.Sent = sentNow(ctx, inv, sent)
|
||||
p.Note = fmt.Sprintf("sent %s %d build(s) that waited for a gate; judging them there", node, len(moves))
|
||||
if said := grantsSaid(grants); said != "" {
|
||||
p.Note += "; " + said
|
||||
}
|
||||
if said := recreationsSaid(moves); said != "" {
|
||||
p.Note += "; " + said
|
||||
}
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
return keptSent, nil
|
||||
}
|
||||
if !alreadySent(ctx, inv, node, g.Carried) {
|
||||
return keptNothing, nil
|
||||
}
|
||||
since := now
|
||||
if s, found, err := inv.SentTo(ctx, node); err == nil && found {
|
||||
if !s.At.IsZero() {
|
||||
since = s.At
|
||||
}
|
||||
g.Sent = map[string]inventory.SentDeclaration{node: s}
|
||||
}
|
||||
g.Since, g.Pending = &since, false
|
||||
p.Note = fmt.Sprintf("%s was already sent the %d build(s) kept for it, at %s: judged from that send, nothing sent again",
|
||||
node, len(g.Carried), since.Local().Format("15:04:05"))
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
return keptAlready, nil
|
||||
}
|
||||
|
||||
// alreadySent says a machine's last send carries every build of the moves: a send made and not kept.
|
||||
func alreadySent(ctx context.Context, inv *inventory.Inventory, node string, moves []inventory.CarriedMove) bool {
|
||||
if len(moves) == 0 {
|
||||
return false
|
||||
}
|
||||
sent, known, err := inv.SentBuilds(ctx, node)
|
||||
if err != nil || !known {
|
||||
return false
|
||||
}
|
||||
for _, mv := range moves {
|
||||
if !sameCommit(sent[mv.Module], mv.To) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// advanceRest takes one step of a release plan's rest (novox/hq ADR 0353): the sends kept for a step made, every
|
||||
// machine of the step judged by its own gate, one that fails put back by itself while the others go on to their
|
||||
// verdicts, and the next step — or the end — once every gate of the step has its verdict or is silent.
|
||||
func advanceRest(ctx context.Context, open *stores, p *inventory.Plan, now time.Time) (bool, error) {
|
||||
r := p.Release
|
||||
// The sends kept for this step, in the plan's order: the control node, last in it, last (§1).
|
||||
var pending []string
|
||||
for _, n := range r.Order {
|
||||
if g := r.Gates[n]; g != nil && g.Pending {
|
||||
pending = append(pending, n)
|
||||
}
|
||||
}
|
||||
if len(pending) > 0 {
|
||||
var sent, already, waiting []string
|
||||
for _, node := range pending {
|
||||
out, err := sendKept(ctx, open, p, node, r.Gates[node], now)
|
||||
if err != nil {
|
||||
return true, err
|
||||
}
|
||||
switch out {
|
||||
case keptSent:
|
||||
sent = append(sent, node)
|
||||
case keptAlready:
|
||||
already = append(already, node)
|
||||
case keptNothing:
|
||||
delete(r.Gates, node)
|
||||
r.Done = append(r.Done, node)
|
||||
case keptWaiting:
|
||||
waiting = append(waiting, node)
|
||||
}
|
||||
}
|
||||
var said []string
|
||||
if len(sent) > 0 {
|
||||
said = append(said, "sent "+strings.Join(sent, ", ")+" at once, each judged there")
|
||||
}
|
||||
if len(already) > 0 {
|
||||
said = append(said, strings.Join(already, ", ")+" already sent it: judged from that send")
|
||||
}
|
||||
if len(waiting) > 0 {
|
||||
said = append(said, "waiting before "+strings.Join(waiting, ", ")+": another walk judges a build it carries")
|
||||
}
|
||||
if len(said) > 0 {
|
||||
p.Note = strings.Join(said, "; ")
|
||||
}
|
||||
return len(sent)+len(already) > 0 || len(waiting) < len(pending), nil
|
||||
}
|
||||
reports, err := open.inventory.LastReports(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
byNode := map[string]inventory.Reported{}
|
||||
for _, rep := range reports {
|
||||
byNode[rep.Node] = rep
|
||||
}
|
||||
changed, noteWas := false, p.Note
|
||||
var notes, judging []string
|
||||
batched, back := batchingRollbacks(ctx)
|
||||
for _, node := range r.Order {
|
||||
g := r.Gates[node]
|
||||
if g == nil || g.Verdict != "" || g.Silent {
|
||||
continue
|
||||
}
|
||||
// What another machine's failure already put back is not this gate's to judge (ADR 0353 §3).
|
||||
if len(g.Carried) > 0 && len(g.Returned) >= len(g.Carried) {
|
||||
decide(g, inventory.GateFailed, "what it was sent was put back with another machine's failure", now)
|
||||
r.Failed = append(r.Failed, node)
|
||||
changed = true
|
||||
continue
|
||||
}
|
||||
// **A machine that says nothing on its send is left silent, not failed** (ADR 0353 §4): it is not judged
|
||||
// until it reports, and past ApplySilentAfter it is listed and left.
|
||||
if !reportedOnSend(byNode[node], node, g) {
|
||||
if now.Sub(*g.Since) > inventory.ApplySilentAfter {
|
||||
g.Silent = true
|
||||
r.Silent = append(r.Silent, node)
|
||||
changed = true
|
||||
notes = append(notes, fmt.Sprintf("%s said nothing on its send within %s: left silent", node,
|
||||
inventory.ApplySilentAfter))
|
||||
continue
|
||||
}
|
||||
judging = append(judging, node+": not yet reported on its send")
|
||||
continue
|
||||
}
|
||||
verdict, err := judgeMoves(ctx, open, g, nil, now)
|
||||
if err != nil {
|
||||
return changed, err
|
||||
}
|
||||
switch verdict {
|
||||
case "":
|
||||
if putBackBroken(ctx, open, p, g, "", nil) {
|
||||
changed = true
|
||||
notes = append(notes, p.Note)
|
||||
}
|
||||
judging = append(judging, node+": "+gateLine(g))
|
||||
case inventory.GatePassed:
|
||||
passCarried(ctx, open, p, g, "")
|
||||
r.Done = append(r.Done, node)
|
||||
changed = true
|
||||
case inventory.GateSuperseded:
|
||||
// Another send moved a judged module there (novox/hq issue 352): nothing judged, nothing put back on
|
||||
// it; what still waits is released again.
|
||||
r.Done = append(r.Done, node)
|
||||
changed = true
|
||||
notes = append(notes, fmt.Sprintf("superseded on %s: %s", node, g.Why))
|
||||
default:
|
||||
// **Put back by itself** (ADR 0353 §3): what was found wanting goes back at once, and every other
|
||||
// machine of the step goes on to its own verdict.
|
||||
stateWas := p.State
|
||||
p.Note = ""
|
||||
failCarried(batched, open, p, g, "")
|
||||
p.State = stateWas
|
||||
r.Failed = append(r.Failed, node)
|
||||
changed = true
|
||||
notes = append(notes, fmt.Sprintf("failed its gate on %s: %s — %s", node, g.Why, p.Note))
|
||||
returnedElsewhere(ctx, open, r, node, g)
|
||||
}
|
||||
}
|
||||
p.Note = ""
|
||||
sendRollbacks(ctx, open, p, back)
|
||||
if p.Note != "" {
|
||||
notes = append(notes, strings.TrimPrefix(p.Note, "; "))
|
||||
}
|
||||
if len(judging) > 0 {
|
||||
note := strings.Join(append(notes, "judging "+strings.Join(judging, "; ")), "; ")
|
||||
p.Note = note
|
||||
return changed || noteWas != note, nil
|
||||
}
|
||||
// Every gate of the step has its verdict or is silent: the next step, unless a machine failed (§3).
|
||||
if len(r.Failed) == 0 {
|
||||
opened, waiting, err := openRestStep(ctx, open, p)
|
||||
if err != nil {
|
||||
return changed, err
|
||||
}
|
||||
if opened {
|
||||
return true, nil
|
||||
}
|
||||
if waiting != "" {
|
||||
note := strings.Join(append(notes, waiting), "; ")
|
||||
p.Note = note
|
||||
return changed || noteWas != note, nil
|
||||
}
|
||||
}
|
||||
p.Tier = len(p.Tiers)
|
||||
end := fmt.Sprintf("released on %s", orNone(strings.Join(r.Done, ", ")))
|
||||
if len(r.Silent) > 0 {
|
||||
end += "; silent on their send, left: " + strings.Join(r.Silent, ", ")
|
||||
}
|
||||
if len(r.Skipped) > 0 {
|
||||
end += "; not heard from, left as they were: " + strings.Join(r.Skipped, ", ")
|
||||
}
|
||||
p.State = inventory.PlanDone
|
||||
if len(r.Failed) > 0 {
|
||||
p.State = inventory.PlanFailed
|
||||
end = "failed on " + strings.Join(r.Failed, ", ") + "; " + end
|
||||
}
|
||||
p.Note = strings.Join(append(notes, end), "; ")
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// reportedOnSend says a machine's last report is about the send its gate judges (novox/hq issue 352): the
|
||||
// declaration kept on the gate, or one sequenced after it; a gate that kept none reads the send made last.
|
||||
func reportedOnSend(r inventory.Reported, node string, g *inventory.PlanGate) bool {
|
||||
if r.At == nil {
|
||||
return false
|
||||
}
|
||||
if s, kept := g.Sent[node]; kept {
|
||||
return s.ReportsOn(r)
|
||||
}
|
||||
return r.Current
|
||||
}
|
||||
|
||||
// returnedElsewhere marks, on every other gate of the rest still judging, the builds a failed machine's
|
||||
// rollback put back wherever they were sent (ADR 0236 §5): no longer theirs to judge (ADR 0353 §3).
|
||||
func returnedElsewhere(ctx context.Context, open *stores, r *inventory.PlanRelease, failed string, g *inventory.PlanGate) {
|
||||
for _, c := range g.Carried {
|
||||
if c.Build == "" {
|
||||
continue
|
||||
}
|
||||
if marked, err := open.inventory.GateFailed(ctx, c.Build); err != nil || !marked {
|
||||
continue
|
||||
}
|
||||
for node, other := range r.Gates {
|
||||
if node == failed || other.Verdict != "" || slices.Contains(other.Returned, c.Module) {
|
||||
continue
|
||||
}
|
||||
for _, oc := range other.Carried {
|
||||
if oc.Module == c.Module && sameCommit(oc.To, c.To) {
|
||||
other.Returned = append(other.Returned, c.Module)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// openRestStep chooses the next step of a release plan's rest and keeps its sends (novox/hq ADR 0353 §1, §2):
|
||||
// every machine left that is heard from and has something waiting, except that a build no gate has passed goes
|
||||
// to one machine of the step only. Answers whether a step was kept, or why the walk waits.
|
||||
func openRestStep(ctx context.Context, open *stores, p *inventory.Plan) (bool, string, error) {
|
||||
r := p.Release
|
||||
heard, err := releaseHeard(ctx, open)
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
f, err := readMoveFacts(ctx, open.inventory)
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
waiting := map[string][]inventory.CarriedMove{}
|
||||
var nodes, walked []string
|
||||
for _, node := range r.Order[min(r.Next, len(r.Order)):] {
|
||||
if r.Gates[node] != nil || slices.Contains(r.Done, node) || slices.Contains(r.Skipped, node) ||
|
||||
slices.Contains(r.Failed, node) || slices.Contains(r.Silent, node) {
|
||||
continue
|
||||
}
|
||||
if !heard[node] {
|
||||
r.Skipped = append(r.Skipped, node)
|
||||
continue
|
||||
}
|
||||
moves, err := gatedMoves(ctx, open, node, nil)
|
||||
if errors.Is(err, errWalkedElsewhere) {
|
||||
walked = append(walked, fmt.Sprintf("%s (%v)", node, err))
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return false, "", fmt.Errorf("reading what waits on %s: %w", node, err)
|
||||
}
|
||||
if len(moves) == 0 {
|
||||
r.Done = append(r.Done, node)
|
||||
continue
|
||||
}
|
||||
waiting[node] = moves
|
||||
nodes = append(nodes, node)
|
||||
}
|
||||
step := restStep(nodes, waiting, f.gated)
|
||||
if len(step) == 0 {
|
||||
if len(walked) > 0 {
|
||||
return false, "waiting before " + strings.Join(walked, "; "), nil
|
||||
}
|
||||
return false, "", nil
|
||||
}
|
||||
if r.Gates == nil {
|
||||
r.Gates = map[string]*inventory.PlanGate{}
|
||||
}
|
||||
for _, node := range step {
|
||||
r.Gates[node] = &inventory.PlanGate{Machines: []string{node}, Carried: waiting[node], Pending: true}
|
||||
}
|
||||
p.Note = fmt.Sprintf("sending %s at once, each judged there", strings.Join(step, ", "))
|
||||
if left := len(nodes) - len(step); left > 0 {
|
||||
p.Note += fmt.Sprintf("; %d machine(s) wait for a build that passed no gate yet to pass on one", left)
|
||||
}
|
||||
return true, "", nil
|
||||
}
|
||||
|
||||
// restStep is the machines of one step of the rest (novox/hq ADR 0353 §2), in the order given: each machine, unless
|
||||
// a build it would carry has passed no gate and an earlier machine of the step already carries it — that build's
|
||||
// first machine. Pure.
|
||||
func restStep(nodes []string, waiting map[string][]inventory.CarriedMove, passed func(module, commit string) bool) []string {
|
||||
first := map[string]string{}
|
||||
var out []string
|
||||
for _, n := range nodes {
|
||||
var unpassed []string
|
||||
for _, mv := range waiting[n] {
|
||||
if !passed(mv.Module, mv.To) {
|
||||
unpassed = append(unpassed, mv.Module+"@"+mv.To)
|
||||
}
|
||||
}
|
||||
if slices.ContainsFunc(unpassed, func(k string) bool { _, taken := first[k]; return taken }) {
|
||||
continue
|
||||
}
|
||||
for _, k := range unpassed {
|
||||
first[k] = n
|
||||
}
|
||||
out = append(out, n)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// backlogObservation is what the gate's probe says of a release held after a failure.
|
||||
func backlogObservation() []conditions.Observation {
|
||||
if backlogNow.held == "" || len(backlogNow.waiting) == 0 {
|
||||
|
||||
@@ -595,7 +595,7 @@ func advancePlans(ctx context.Context, open *stores) {
|
||||
return
|
||||
}
|
||||
defer release()
|
||||
// What waits for a gate, released one machine at a time (ADR 0236).
|
||||
// What waits for a gate, released one machine first, then the rest at once (ADR 0236, ADR 0353).
|
||||
if _, err := releaseBacklog(ctx, open, ""); err != nil {
|
||||
fmt.Printf("plans: what waits for a gate could not be looked at: %v\n", err)
|
||||
}
|
||||
@@ -1056,6 +1056,15 @@ func firstSend(ctx context.Context, open *stores, p *inventory.Plan, node string
|
||||
runningOf map[string][]string) error {
|
||||
inv := open.inventory
|
||||
before, beforeKnown, _ := inv.SentBuilds(ctx, node)
|
||||
// **A machine already sent what this send is for is not sent again** (novox/hq ADR 0353 §6): sent by this plan
|
||||
// before a controller was replaced and before it was kept, or by another walk. Judged from that send, and from
|
||||
// the readings of a gate already judging it there.
|
||||
if beforeKnown && slices.IndexFunc(modules, func(m string) bool { return !sameCommit(before[m], p.Modules[m].Commit) }) < 0 {
|
||||
if others, err := gatedMoves(ctx, open, node, nil); err == nil && len(others) == 0 {
|
||||
adoptSend(ctx, open, p, node, modules, runningOf)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
owns := make([]inventory.CarriedMove, 0, len(modules))
|
||||
for _, m := range modules {
|
||||
s := p.Modules[m]
|
||||
@@ -1106,6 +1115,86 @@ func firstSend(ctx context.Context, open *stores, p *inventory.Plan, node string
|
||||
return nil
|
||||
}
|
||||
|
||||
// adoptSend records a first send the machine already had (novox/hq ADR 0353 §6, §7): nothing is sent; the gate counts
|
||||
// from the machine's last send, and takes over the readings of an open gate of another walk judging the same build
|
||||
// there — whose carried move also says what the machine ran before, for a rollback.
|
||||
func adoptSend(ctx context.Context, open *stores, p *inventory.Plan, node string, modules []string,
|
||||
runningOf map[string][]string) {
|
||||
inv := open.inventory
|
||||
since := time.Now().UTC()
|
||||
last, found, err := inv.SentTo(ctx, node)
|
||||
if err == nil && found && !last.At.IsZero() {
|
||||
since = last.At
|
||||
}
|
||||
lead := modules[0]
|
||||
other, id, from := judgingThere(ctx, inv, p.ID, lead, node, p.Modules[lead].Commit)
|
||||
for _, m := range modules {
|
||||
s := p.Modules[m]
|
||||
if m == lead && from != "" {
|
||||
s.Previous = from
|
||||
}
|
||||
at := since
|
||||
s.First, s.FirstAt = []string{node}, &at
|
||||
s.Gate = &inventory.PlanGate{Component: coreComponent(m), Machines: firstRunning([]string{node}, runningOf[m]),
|
||||
From: s.Previous, To: s.Commit, Since: &at}
|
||||
if found {
|
||||
s.Gate.Sent = map[string]inventory.SentDeclaration{node: last}
|
||||
}
|
||||
s.GatedBy = ""
|
||||
if m == lead {
|
||||
s.Gate.TakeReadings(other, modules, id)
|
||||
} else {
|
||||
s.GatedBy = lead
|
||||
}
|
||||
}
|
||||
p.State = inventory.PlanRolling
|
||||
p.Note = fmt.Sprintf("tier %d built; %s already runs %s's build: nothing sent again, judged from that send",
|
||||
p.Tier, node, strings.Join(modules, ", "))
|
||||
if id != "" {
|
||||
p.Note += ", going on from the readings of " + id
|
||||
}
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
}
|
||||
|
||||
// judgingThere is an open gate of another walk judging a module's build on a machine (novox/hq ADR 0353 §7), the
|
||||
// walk's id, and what the machine ran before; nil when none does.
|
||||
func judgingThere(ctx context.Context, inv *inventory.Inventory, self, module, node, commit string) (*inventory.PlanGate, string, string) {
|
||||
plans, err := inv.OpenPlans(ctx)
|
||||
if err != nil {
|
||||
return nil, "", ""
|
||||
}
|
||||
usable := func(g *inventory.PlanGate) bool {
|
||||
return g != nil && !g.Pending && g.Since != nil && (g.Verdict == "" || g.Verdict == inventory.GatePassed) &&
|
||||
slices.Contains(g.Machines, node)
|
||||
}
|
||||
for _, q := range plans {
|
||||
if q.ID == self {
|
||||
continue
|
||||
}
|
||||
if r := q.Release; r != nil {
|
||||
gates := []*inventory.PlanGate{r.Gate}
|
||||
for _, g := range r.Gates {
|
||||
gates = append(gates, g)
|
||||
}
|
||||
for _, g := range gates {
|
||||
if !usable(g) {
|
||||
continue
|
||||
}
|
||||
for _, c := range g.Carried {
|
||||
if c.Module == module && c.Node == node && sameCommit(c.To, commit) {
|
||||
return g, q.ID, c.From
|
||||
}
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
if s := q.Modules[module]; s != nil && usable(s.Gate) && sameCommit(s.Commit, commit) {
|
||||
return s.Gate, q.ID, s.Previous
|
||||
}
|
||||
}
|
||||
return nil, "", ""
|
||||
}
|
||||
|
||||
// passedWith keeps, on a module sent with others, the verdict of the gate that judged the send: the gate
|
||||
// on the first of them passed, and its pass was kept for every build it carried (passCarried).
|
||||
func passedWith(module string, s *inventory.PlanModule, lead string, g *inventory.PlanGate) {
|
||||
@@ -1594,9 +1683,19 @@ func plansCommand(ctx context.Context, args []string) error {
|
||||
// A release plan's walk (ADR 0236): machines done, the one judged, those to come.
|
||||
fmt.Printf(" machines in order: %s; done: %s; skipped: %s\n", strings.Join(r.Order, ", "),
|
||||
orNone(strings.Join(r.Done, ", ")), orNone(strings.Join(r.Skipped, ", ")))
|
||||
if r.Gate != nil {
|
||||
fmt.Printf(" %s\n", gateLine(r.Gate))
|
||||
for _, c := range r.Gate.Carried {
|
||||
if len(r.Failed)+len(r.Silent) > 0 {
|
||||
fmt.Printf(" failed: %s; silent: %s\n", orNone(strings.Join(r.Failed, ", ")), orNone(strings.Join(r.Silent, ", ")))
|
||||
}
|
||||
gates := []*inventory.PlanGate{r.Gate}
|
||||
for _, n := range r.Order {
|
||||
gates = append(gates, r.Gates[n])
|
||||
}
|
||||
for _, g := range gates {
|
||||
if g == nil {
|
||||
continue
|
||||
}
|
||||
fmt.Printf(" %s\n", gateLine(g))
|
||||
for _, c := range g.Carried {
|
||||
fmt.Printf(" %-22s %s → %s\n", c.Module, short(c.From), short(c.To))
|
||||
if c.Recreates != "" {
|
||||
fmt.Printf(" %-22s %s\n", "", c.Recreates)
|
||||
|
||||
@@ -79,7 +79,7 @@ func (f following) Upgraded(ctx context.Context, u link.Upgraded) error {
|
||||
// `rebuild`, a `replay` registered — was sent here one machine after another without any judging; it
|
||||
// now waits for a gate like any other build, and the release plan walks it through the machines, one
|
||||
// at a time, each judged.
|
||||
fmt.Printf("%s moved to %s outside a plan; a release plan sends it to %s, one machine at a time, each judged "+
|
||||
fmt.Printf("%s moved to %s outside a plan; a release plan sends it to %s, one machine first and then the rest at once, each judged "+
|
||||
"(`upgrade backlog` lists what waits)\n", u.Module, shortCommit(u.Commit), readableList(on))
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -177,6 +177,12 @@ func walkReached(p inventory.Plan, running func(string) []string) map[string][]s
|
||||
reached = appendNew(reached, r.Gate.Machines...)
|
||||
carried(r.Gate)
|
||||
}
|
||||
for _, g := range r.Gates {
|
||||
if g != nil && !g.Pending {
|
||||
reached = appendNew(reached, g.Machines...)
|
||||
carried(g)
|
||||
}
|
||||
}
|
||||
for _, tier := range p.Tiers {
|
||||
for _, m := range tier {
|
||||
for _, n := range running(m) {
|
||||
|
||||
Reference in New Issue
Block a user