checker: claim no check of a repository an approved merge holds a turn on, and claim it when the turn goes (ADR 0548 §3; issue 568 D1)
This commit is contained in:
@@ -174,7 +174,9 @@ func Tools(r check.Reader) []stdio.Tool {
|
||||
"and the candidates whose check is owed by kind (waiting_by_kind: new, nodes, and trunk, a re-judge pulled after the other two). A key a " +
|
||||
"build module left running (started past the step bound, or its build module not running on its node) is listed apart under " +
|
||||
"left_running with why, and not counted. Nothing holds the checker and it has no capacity of its own: a check is an ask on the " +
|
||||
"build seats' streams, whose backlog the delivery view shows (novox/hq ADR 0525). Read-only.",
|
||||
"build seats' streams, whose backlog the delivery view shows (novox/hq ADR 0525). held_by_turn names the candidates whose check an approved " +
|
||||
"merge's turn holds off (novox/hq ADR 0548 §3), with the repository and the merge that waits; the gate's turns tool reads every turn and its " +
|
||||
"hand-now tool steers one. Read-only.",
|
||||
Input: map[string]any{"type": "object", "properties": map[string]any{}},
|
||||
Run: func(map[string]any) (any, error) { return r.Status() }},
|
||||
}
|
||||
@@ -193,7 +195,8 @@ func main() {
|
||||
r := check.Reader{VerdictState: read(stdio.State("verdicts")), KeptState: read(stdio.State("kept")), CandidateState: read(stdio.State("mesh-delivery-indexer.candidates")),
|
||||
PreviewState: read(stdio.State("previews")),
|
||||
StepStates: map[string]check.StateRead{"mesh-build-go": read(stdio.State("mesh-build-go.steps")), "mesh-build-node": read(stdio.State("mesh-build-node.steps"))},
|
||||
HaveState: read(stdio.State("node-engine.have")), StepBound: stepBound()}
|
||||
HaveState: read(stdio.State("node-engine.have")), StepBound: stepBound(),
|
||||
TurnState: read(stdio.State("mesh-delivery-gate.turns"))}
|
||||
tries := stdio.State(check.TriesState)
|
||||
source := stdio.State("mesh-delivery-gate.source")
|
||||
trunks := func() (map[string]string, error) {
|
||||
|
||||
@@ -193,6 +193,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
tools := &buildrun.ToolchainSet{}
|
||||
presence := &buildrun.PresenceSet{}
|
||||
trunks := &check.Trunks{}
|
||||
turns := &check.Turns{}
|
||||
// A check is one ask per pull request and build module on the build seats' check stream, pulled by a build node with a
|
||||
// free slot (novox/hq ADR 0525 §1, §3): the checker chooses no node and starts no step.
|
||||
asks := buildrun.JetStreamAsks{JS: js}
|
||||
@@ -200,6 +201,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]},
|
||||
Owed: kvStore{buckets[delivery.CheckerOwedBucket]},
|
||||
Trunks: trunks,
|
||||
Turns: turns,
|
||||
Bound: checkBound,
|
||||
Link: link,
|
||||
Logf: logger.Printf,
|
||||
@@ -313,6 +315,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}()
|
||||
}
|
||||
}
|
||||
// The gate's merge turns (novox/hq ADR 0548 §3): no check is claimed of a repository an approved merge waits on, and the
|
||||
// delete of its turn claims it. Read before the candidates, so no check starts past a turn written before this start.
|
||||
// Where the gate's bucket is not there yet it is read as empty: no turn holds a check until the checker starts again.
|
||||
if kv := process.OptionalBucket(ctx, js, delivery.TurnsBucket, "no merge turn holds a check until the checker starts again", logger.Printf); kv != nil {
|
||||
tw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { turns.Put(e.Key(), e.Value(), !opOf(e)) })
|
||||
if err != nil {
|
||||
process.NotWatched(logger.Printf, delivery.TurnsBucket, err, "no merge turn holds a check until the checker starts again")
|
||||
} else {
|
||||
go func() {
|
||||
errs <- follow(ctx, tw, func(e jetstream.KeyValueEntry) {
|
||||
if turns.Put(e.Key(), e.Value(), !opOf(e)) {
|
||||
ch.Reconsider(ctx) // a turn went or changed: a check it held is claimed now
|
||||
}
|
||||
})
|
||||
}()
|
||||
}
|
||||
}
|
||||
// The modules say which repository holds a TypeScript module, so the Node build module judges it too.
|
||||
mw, err := settle(ctx, buckets[delivery.ModulesBucket], func(e jetstream.KeyValueEntry) { repos.Put(ctx, e.Key(), e.Value(), !opOf(e)) })
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,114 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// One check per delivery (novox/hq ADR 0548): the checker's part. A merge turn holds new checks off its repository (§3);
|
||||
// a trunk or sibling move never ends a running check and carries its verdict once it ended (§1, §5, §6); only a head git
|
||||
// refused gets a whole check after a move (§7); the line of a carried verdict says so (§8).
|
||||
|
||||
// turnOf is a turn's value on a repository for a delivery of the heads hash.
|
||||
func turnOf(t testing.TB, repo, group, heads string) []byte {
|
||||
t.Helper()
|
||||
at := time.Date(2026, 10, 26, 9, 0, 0, 0, time.UTC)
|
||||
data, err := delivery.EncodeTurn(delivery.Turn{Repository: repo, Waiting: []delivery.TurnWait{{Group: group, Heads: heads, Approved: at, Since: at}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return data
|
||||
}
|
||||
|
||||
func turnKey(t testing.TB, repo string) string {
|
||||
t.Helper()
|
||||
k, err := delivery.TurnsKey(repo)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return k
|
||||
}
|
||||
|
||||
// waitFor polls a condition the checker's goroutines make true, failing after two seconds.
|
||||
func waitFor(t testing.TB, what string, ok func() bool) {
|
||||
t.Helper()
|
||||
for deadline := time.Now().Add(2 * time.Second); !ok(); time.Sleep(time.Millisecond) {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("%s did not happen", what)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestOneCheckPerDelivery(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
for name, run := range map[string]func(t *testing.T){
|
||||
// §3: no claim while a turn names one of its repositories; a claim on the turn's delete event.
|
||||
"a turn on its repository holds a new check off until the turn goes": func(t *testing.T) {
|
||||
j := passJudge()
|
||||
r := newRig(t, j)
|
||||
turns := &Turns{}
|
||||
r.ch.Turns = turns
|
||||
turns.Put(turnKey(t, repoA), turnOf(t, repoA, "fix/other", "0123456789ab"), false)
|
||||
cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1))
|
||||
r.ch.Candidate(ctx, "feat%2Fx", cand)
|
||||
r.ch.Wait()
|
||||
if j.count() != 0 || len(r.verdicts.vals) != 0 {
|
||||
t.Fatalf("judged %d times with a turn on %s, verdicts %v", j.count(), repoA, r.verdicts.vals)
|
||||
}
|
||||
if held := r.ch.Held()[keyOf(t, "feat/x", cand)]; !strings.Contains(held, repoA) || !strings.Contains(held, "fix/other") {
|
||||
t.Errorf("the hold is said as %q, want the turn's repository and group", held)
|
||||
}
|
||||
if !turns.Put(turnKey(t, repoA), nil, true) {
|
||||
t.Fatal("the turn's delete changed nothing")
|
||||
}
|
||||
r.ch.Reconsider(ctx) // what the watch of turns does on the delete
|
||||
r.ch.Wait()
|
||||
if j.count() != 1 || r.verdict(t, keyOf(t, "feat/x", cand)).State != delivery.VerdictPass {
|
||||
t.Errorf("judged %d times after the turn went, want once", j.count())
|
||||
}
|
||||
},
|
||||
"a turn on another repository, or of its own heads, holds nothing": func(t *testing.T) {
|
||||
j := passJudge()
|
||||
r := newRig(t, j)
|
||||
turns := &Turns{}
|
||||
r.ch.Turns = turns
|
||||
cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1))
|
||||
c, _ := delivery.DecodeCandidate(cand)
|
||||
turns.Put(turnKey(t, repoSDK), turnOf(t, repoSDK, "fix/other", "0123456789ab"), false)
|
||||
turns.Put(turnKey(t, repoA), turnOf(t, repoA, "feat/x", GateHeads(c)), false)
|
||||
r.ch.Candidate(ctx, "feat%2Fx", cand)
|
||||
r.ch.Wait()
|
||||
if j.count() != 1 {
|
||||
t.Errorf("judged %d times, want once: the turns hold another repository and its own merge", j.count())
|
||||
}
|
||||
},
|
||||
// §3, the race: a check claimed before the gate's turn is judged on the commit it read, never restarted.
|
||||
"a check claimed before a turn was written runs to its verdict": func(t *testing.T) {
|
||||
release, started := make(chan struct{}), make(chan struct{})
|
||||
j := &judgeFunc{f: func(context.Context, *Case, int) (Outcome, error) {
|
||||
close(started)
|
||||
<-release
|
||||
return Outcome{Verdict: passing()}, nil
|
||||
}}
|
||||
r := newRig(t, j)
|
||||
turns := &Turns{}
|
||||
r.ch.Turns = turns
|
||||
cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1))
|
||||
r.ch.Candidate(ctx, "feat%2Fx", cand)
|
||||
<-started
|
||||
turns.Put(turnKey(t, repoA), turnOf(t, repoA, "fix/other", "0123456789ab"), false)
|
||||
r.ch.Reconsider(ctx)
|
||||
close(release)
|
||||
r.ch.Wait()
|
||||
if j.count() != 1 || r.verdict(t, keyOf(t, "feat/x", cand)).State != delivery.VerdictPass {
|
||||
t.Errorf("judged %d times; the running check is not ended by the turn", j.count())
|
||||
}
|
||||
},
|
||||
} {
|
||||
t.Run(name, run)
|
||||
}
|
||||
}
|
||||
@@ -65,6 +65,11 @@ type Checker struct {
|
||||
Owed KV
|
||||
Judge Judge
|
||||
Trunks interface{ Heads() map[string]string }
|
||||
// Turns are the gate's merge turns (ADR 0548 §3): no check is claimed of a repository a turn names, but the check of the
|
||||
// waiting delivery's own heads; nil holds none.
|
||||
Turns interface {
|
||||
Holding(repositories []string, heads string) (string, delivery.TurnWait, bool)
|
||||
}
|
||||
// Bound is how long one check may run; none when 0 (DefaultBound).
|
||||
Bound time.Duration
|
||||
// Link is the base of the address a status points at; the verdict key follows.
|
||||
@@ -83,12 +88,13 @@ type Checker struct {
|
||||
tried map[string]bool // by verdict key: looked at since this process started
|
||||
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
|
||||
}
|
||||
|
||||
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 = map[string]uint64{}
|
||||
ch.nodesFail, ch.held = map[string]uint64{}, map[string]string{}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -265,6 +271,9 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64
|
||||
return nil, fmt.Errorf("%s: %w", cs.Key, err)
|
||||
}
|
||||
if !found {
|
||||
if ch.heldByTurn(cs) {
|
||||
return nil, nil
|
||||
}
|
||||
rev, err := ch.Verdicts.Create(ctx, cs.Key, data)
|
||||
if errors.Is(err, ErrConflict) {
|
||||
return nil, nil // another checker made it first
|
||||
@@ -293,7 +302,7 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64
|
||||
ch.mu.Unlock()
|
||||
retake = !ok || epoch > at
|
||||
}
|
||||
if !retake {
|
||||
if !retake || ch.heldByTurn(cs) {
|
||||
return nil, nil
|
||||
}
|
||||
rev, err = ch.Verdicts.Update(ctx, cs.Key, data, rev)
|
||||
@@ -306,6 +315,50 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64
|
||||
return &claimed{rev}, nil
|
||||
}
|
||||
|
||||
// heldByTurn says whether a merge turn holds the case's check off (ADR 0548 §3): an approved merge waits on one of the
|
||||
// repositories it merges heads into, and it is not the merge of the case's own heads. The check is claimed when the turn's
|
||||
// key goes (Reconsider on the turns watch). Said once per turn in the log. A check already claimed is never held: one
|
||||
// claimed in the moment before the gate wrote its turn is judged on the commit it read and carried once after.
|
||||
func (ch *Checker) heldByTurn(cs *Case) bool {
|
||||
if ch.Turns == nil {
|
||||
return false
|
||||
}
|
||||
repos := make([]string, 0, len(cs.Trunk))
|
||||
for r := range cs.Trunk {
|
||||
repos = append(repos, r)
|
||||
}
|
||||
repo, w, held := ch.Turns.Holding(repos, cs.Heads)
|
||||
why := ""
|
||||
if held {
|
||||
why = repo + " " + w.Group + " " + w.Heads
|
||||
}
|
||||
ch.mu.Lock()
|
||||
said := ch.held[cs.Key]
|
||||
if why == "" {
|
||||
delete(ch.held, cs.Key)
|
||||
} else {
|
||||
ch.held[cs.Key] = why
|
||||
}
|
||||
ch.mu.Unlock()
|
||||
if held && said != why {
|
||||
ch.logf("%s: not claimed: the approved merge of %s (heads %s) holds a turn on %s, so its check waits for that merge (ADR 0548 §3)",
|
||||
cs.Key, w.Group, w.Heads, repo)
|
||||
}
|
||||
return held
|
||||
}
|
||||
|
||||
// Held is every check a merge turn holds off now, by verdict key, with the turn's repository, group and heads.
|
||||
func (ch *Checker) Held() map[string]string {
|
||||
ch.init()
|
||||
ch.mu.Lock()
|
||||
defer ch.mu.Unlock()
|
||||
out := make(map[string]string, len(ch.held))
|
||||
for k, v := range ch.held {
|
||||
out[k] = v
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// IsNodesFailure says whether a verdict failed because a part of the mesh the check reads was not heard (UnheardWords):
|
||||
// it is judged again when the nodes change. No check fails for want of a build node any more: its asks wait in the build
|
||||
// seats' streams (ADR 0525 §6, issue 1912).
|
||||
|
||||
@@ -39,7 +39,9 @@ type Reader struct {
|
||||
StepBound time.Duration // how long the checker waits for one step; DefaultStepBound when 0
|
||||
// Trunks are the trunk heads of every repository (the gate's source); nil names no re-judge after a trunk move.
|
||||
Trunks func() (map[string]string, error)
|
||||
Now func() time.Time
|
||||
// TurnState is the gate's merge turns (ADR 0548 §3); without it no check is named as held by a turn.
|
||||
TurnState StateRead
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
// VerdictRead is a verdict as stored with the full report of the status of each pull request it judged (Report), when the
|
||||
@@ -312,7 +314,46 @@ type Status struct {
|
||||
// Waiting are the candidates whose check waits, by kind (ADR 0536 §4): new (no verdict for its heads) and nodes (failed
|
||||
// for the nodes, asked again when they change) start before trunk (judged on trunk heads that moved since).
|
||||
Waiting map[string][]string `json:"waiting_by_kind"`
|
||||
Notes []string `json:"notes,omitempty"`
|
||||
// HeldByTurn are the candidates whose check an approved merge's turn holds off (ADR 0548 §3), by group: the repository and
|
||||
// the group of the merge that waits. Their check is claimed when that turn goes.
|
||||
HeldByTurn map[string]string `json:"held_by_turn"`
|
||||
Notes []string `json:"notes,omitempty"`
|
||||
}
|
||||
|
||||
// turns reads the gate's merge turns, by repository; nil when they cannot be read, said in a note.
|
||||
func (r Reader) turns(st *Status) *Turns {
|
||||
if r.TurnState.Keys == nil || r.TurnState.Get == nil {
|
||||
st.Notes = append(st.Notes, "the gate's merge turns are not readable here, so no check is named as held by one")
|
||||
return nil
|
||||
}
|
||||
keys, err := r.TurnState.Keys()
|
||||
if err != nil {
|
||||
st.Notes = append(st.Notes, "the gate's merge turns cannot be listed, so no check is named as held by one: "+err.Error())
|
||||
return nil
|
||||
}
|
||||
t := &Turns{}
|
||||
for _, k := range keys {
|
||||
if raw, err := r.TurnState.Get(k); err == nil && len(raw) > 0 {
|
||||
t.Put(k, raw, false)
|
||||
}
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
||||
// heldBy says the turn that holds a candidate's check off, or "".
|
||||
func heldBy(t *Turns, c delivery.Candidate, heads map[string]string) string {
|
||||
cs := CaseOf(c, heads)
|
||||
if t == nil || cs == nil {
|
||||
return ""
|
||||
}
|
||||
repos := make([]string, 0, len(cs.Trunk))
|
||||
for r := range cs.Trunk {
|
||||
repos = append(repos, r)
|
||||
}
|
||||
if repo, w, ok := t.Holding(repos, cs.Heads); ok {
|
||||
return repo + ": the approved merge of " + w.Group + " waits on it"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// Status reads the build modules' steps and the candidates against their verdicts.
|
||||
@@ -383,6 +424,8 @@ func (r Reader) Status() (Status, error) {
|
||||
}
|
||||
st.Unjudged = []string{}
|
||||
st.Waiting = map[string][]string{KindNew: {}, KindNodes: {}, KindTrunk: {}}
|
||||
st.HeldByTurn = map[string]string{}
|
||||
turns := r.turns(&st)
|
||||
var heads map[string]string
|
||||
if r.Trunks == nil {
|
||||
st.Notes = append(st.Notes, "the trunk heads are not readable here, so no check is named as waiting for a trunk move")
|
||||
@@ -435,6 +478,9 @@ func (r Reader) Status() (Status, error) {
|
||||
if len(raw) == 0 {
|
||||
st.Unjudged = append(st.Unjudged, c.Group)
|
||||
st.Waiting[KindNew] = append(st.Waiting[KindNew], c.Group)
|
||||
if why := heldBy(turns, c, heads); why != "" {
|
||||
st.HeldByTurn[c.Group] = why
|
||||
}
|
||||
continue
|
||||
}
|
||||
v, err := delivery.DecodeVerdict(raw)
|
||||
@@ -445,6 +491,9 @@ func (r Reader) Status() (Status, error) {
|
||||
case heads != nil:
|
||||
if cs := CaseOf(c, heads); cs != nil && !sameTrunk(v.Trunk, cs.Trunk) {
|
||||
st.Waiting[KindTrunk] = append(st.Waiting[KindTrunk], c.Group)
|
||||
if why := heldBy(turns, c, heads); why != "" {
|
||||
st.HeldByTurn[c.Group] = why
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -234,6 +234,18 @@ func TestTheStatusCountsAnHQOnlyGroupAsUnjudged(t *testing.T) {
|
||||
if len(st.Unjudged) != 1 || st.Unjudged[0] != "pb-x" {
|
||||
t.Errorf("unjudged = %v, want pb-x", st.Unjudged)
|
||||
}
|
||||
// ADR 0548 §3: a candidate whose check an approved merge's turn holds off is named with the turn.
|
||||
held := candidateOf(t, "feat/x", pullOf(repoA, 7, head1))
|
||||
hkey, _ := delivery.CandidatesKey("feat/x")
|
||||
st, err = Reader{CandidateState: fakeState{hkey: held}.read(), VerdictState: fakeState{}.read(),
|
||||
Trunks: func() (map[string]string, error) { return map[string]string{repoA: trunkA}, nil },
|
||||
TurnState: fakeState{turnKey(t, repoA): turnOf(t, repoA, "fix/other", "0123456789ab")}.read()}.Status()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if why := st.HeldByTurn["feat/x"]; !strings.Contains(why, repoA) || !strings.Contains(why, "fix/other") {
|
||||
t.Errorf("held_by_turn = %v, want feat/x held by the merge of fix/other on %s", st.HeldByTurn, repoA)
|
||||
}
|
||||
}
|
||||
|
||||
// slowBucket wraps a state with a Get that takes delay and counts the values read, the lists and the most Gets in flight.
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// Turns holds the gate's merge turns (novox/hq ADR 0548 §3), by repository, as a watch of the gate's turns bucket puts them
|
||||
// in: the checker claims no check of a repository a turn names, except the check of a waiting delivery's own heads, so an
|
||||
// approved merge goes before a check not yet started. A bucket not there is read as empty.
|
||||
type Turns struct {
|
||||
mu sync.Mutex
|
||||
byRepo map[string]delivery.Turn
|
||||
}
|
||||
|
||||
// Put takes a value of the turns bucket in, or its delete, and answers whether what the checker holds changed. A value that
|
||||
// is not a turn is dropped as a delete would be: a turn the checker cannot read holds nothing.
|
||||
func (t *Turns) Put(key string, value []byte, deleted bool) bool {
|
||||
repository, err := delivery.ParseTurnsKey(key)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
if t.byRepo == nil {
|
||||
t.byRepo = map[string]delivery.Turn{}
|
||||
}
|
||||
cur, had := t.byRepo[repository]
|
||||
if !deleted {
|
||||
if turn, err := delivery.DecodeTurn(value); err == nil {
|
||||
t.byRepo[repository] = turn
|
||||
return !had || !delivery.SameTurn(cur, turn)
|
||||
}
|
||||
}
|
||||
delete(t.byRepo, repository)
|
||||
return had
|
||||
}
|
||||
|
||||
// Holding names the first of the repositories (by name) a turn holds off the check of the heads, and the delivery that
|
||||
// waits first in it; false when none does.
|
||||
func (t *Turns) Holding(repositories []string, heads string) (string, delivery.TurnWait, bool) {
|
||||
if t == nil {
|
||||
return "", delivery.TurnWait{}, false
|
||||
}
|
||||
sorted := make([]string, 0, len(repositories))
|
||||
for _, r := range repositories {
|
||||
sorted = append(sorted, strings.ToLower(r))
|
||||
}
|
||||
sort.Strings(sorted)
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
for _, r := range sorted {
|
||||
turn, ok := t.byRepo[r]
|
||||
if !ok || !delivery.TurnHolds(turn, heads) {
|
||||
continue
|
||||
}
|
||||
for _, w := range turn.Waiting {
|
||||
if w.Heads != heads {
|
||||
return r, w, true
|
||||
}
|
||||
}
|
||||
}
|
||||
return "", delivery.TurnWait{}, false
|
||||
}
|
||||
|
||||
// All is every turn held now, by repository (a copy).
|
||||
func (t *Turns) All() map[string]delivery.Turn {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
out := make(map[string]delivery.Turn, len(t.byRepo))
|
||||
for r, turn := range t.byRepo {
|
||||
out[r] = turn
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -97,7 +97,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments",
|
||||
"mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-gate.turns", "mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
|
||||
@@ -25,6 +25,7 @@
|
||||
"mesh-delivery-assigner.compositions",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-gate.source",
|
||||
"mesh-delivery-gate.turns",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
|
||||
Reference in New Issue
Block a user