CK1: the checker reads no merge turn of the gate and claims a new check at once
Builds: design 47 G §278, §197 (test 52) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -422,8 +422,8 @@ func main() {
|
||||
PreviewState: read(stdio.State("previews")), InputState: read(stdio.State(check.InputsStateName)),
|
||||
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(),
|
||||
TurnState: read(stdio.State("mesh-delivery-gate.turns")), HearingState: read(stdio.State(buildrun.HearingState)),
|
||||
AskState: read(stdio.State(check.AsksState)), BuildState: read(stdio.State("mesh-delivery-builder.builds"))}
|
||||
HearingState: read(stdio.State(buildrun.HearingState)),
|
||||
AskState: read(stdio.State(check.AsksState)), BuildState: read(stdio.State("mesh-delivery-builder.builds"))}
|
||||
r.Runs = runsOf{ownerRead(stdio.Ask, indexerModulesTool)} // the node-engine at the commit the mesh runs, as the checker reads it
|
||||
tries := stdio.State(check.TriesState)
|
||||
trunks := func() (map[string]string, error) {
|
||||
|
||||
@@ -38,7 +38,6 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
@@ -608,7 +607,7 @@ func startBed(t *testing.T) *bed {
|
||||
check.CheckerOwedBucket: 5, check.PreviewsBucket: 5, check.CandidatesBucket: 10,
|
||||
check.ModulesBucket: 20, check.ManifestsBucket: 1,
|
||||
check.PresenceBucket: 20, check.HaveBucket: 20, check.TriesBucket: 1,
|
||||
check.TurnsBucket: 5, builds.BuilderBuildsBucket: 1, check.InputsBucket: 1} {
|
||||
builds.BuilderBuildsBucket: 1, check.InputsBucket: 1} {
|
||||
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: uint8(history)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -936,26 +935,8 @@ func TestACheckOrACarryOfEqualInputsRunsNothing(t *testing.T) { // 1287/AC5, 128
|
||||
t.Errorf("%d kept keys, was %d", got, kept)
|
||||
}
|
||||
|
||||
// §3: the gate's turn on the repository holds a new check off; its delete claims it.
|
||||
at := time.Now().UTC()
|
||||
turn, err := meshdeliveryclient.EncodeTurn(meshdelivery.TurnsByRepositoryValue{Repository: repo, Waiting: []meshdelivery.TurnsByRepositoryValueWaitingItem{{Group: "fix/other", Heads: "0123456789ab", Approved: at, Since: at}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tk, _ := meshdeliveryclient.TurnsKey(repo)
|
||||
turnRev, err := b.buckets[check.TurnsBucket].Put(b.ctx, tk, turn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.said(t, fmt.Sprintf("merge turn %s changed (revision %d)", tk, turnRev)) // the checker heard the turn
|
||||
// Design 47 G §278 (test 52): a new check is claimed at once, with no state of the gate granted to the checker.
|
||||
third := b.candidate(t, "feat/z", pull(repo, 9, "3333333333333333333333333333333333333333"))
|
||||
b.said(t, third+": not claimed: the approved merge of fix/other") // it held the check off
|
||||
if _, err := b.buckets[check.CheckerVerdictsBucket].Get(b.ctx, third); !errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
t.Fatalf("verdicts.%s exists (%v) while a merge turn holds its repository", third, err)
|
||||
}
|
||||
if err := b.buckets[check.TurnsBucket].Delete(b.ctx, tk); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.verdict(t, third, ended)
|
||||
|
||||
// §5: the trunk moves; the first verdict is carried by its digests, nothing runs.
|
||||
|
||||
@@ -1,50 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"mesh-delivery-checker/internal/check"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
|
||||
)
|
||||
|
||||
// lazyTurns is the gate's merge turns as the checker reads them (novox/hq ADR 0548 §3), opened when the gate's bucket is
|
||||
// there: before each claim it is opened again until it is, so a checker that started before the gate made the bucket holds
|
||||
// checks from the gate's first turn on, with no timer and no restart. Until then it holds nothing, and says so once, loudly,
|
||||
// and once again when it opens.
|
||||
type lazyTurns struct {
|
||||
turns *check.Turns
|
||||
open func() error
|
||||
logf func(format string, args ...any)
|
||||
|
||||
mu sync.Mutex
|
||||
opened bool
|
||||
said bool
|
||||
}
|
||||
|
||||
// ensure opens the bucket once it is there; a failure is said once.
|
||||
func (l *lazyTurns) ensure() {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
if l.opened || l.open == nil {
|
||||
return
|
||||
}
|
||||
if err := l.open(); err != nil {
|
||||
if !l.said {
|
||||
l.said = true
|
||||
l.logf("NO MERGE TURN HOLDS A CHECK: the gate's bucket %s cannot be read (%v); every check is claimed as if no approved merge waited, "+
|
||||
"and the bucket is opened again before each claim until it is there", check.TurnsBucket, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
l.opened = true
|
||||
if l.said {
|
||||
l.logf("the gate's merge turns are read from %s from now on: an approved merge's turn holds checks off its repositories again", check.TurnsBucket)
|
||||
}
|
||||
}
|
||||
|
||||
// Holding opens the bucket when it is not yet, then answers what the turns hold.
|
||||
func (l *lazyTurns) Holding(repositories []string, heads string) (string, meshdelivery.TurnsByRepositoryValueWaitingItem, bool) {
|
||||
l.ensure()
|
||||
return l.turns.Holding(repositories, heads)
|
||||
}
|
||||
@@ -204,8 +204,6 @@ 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{}
|
||||
lazy := &lazyTurns{turns: turns, logf: logger.Printf}
|
||||
// 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}
|
||||
@@ -232,7 +230,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
Verdicts: kvStore{buckets[check.CheckerVerdictsBucket]},
|
||||
Owed: kvStore{buckets[check.CheckerOwedBucket]},
|
||||
Trunks: trunks,
|
||||
Turns: lazy,
|
||||
Bound: checkBound,
|
||||
Link: link,
|
||||
Logf: logger.Printf,
|
||||
@@ -381,31 +378,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return fmt.Errorf("the checker cannot hear the catalog half's revisions: %w", err)
|
||||
}
|
||||
firstTrunks(ctx)
|
||||
// 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, and opened again before each claim until it is there
|
||||
// (lazyTurns), so a gate delivered after the checker started holds checks from its first turn on.
|
||||
lazy.open = func() error {
|
||||
kv, err := js.KeyValue(ctx, check.TurnsBucket)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { turns.Put(e.Key(), e.Value(), !opOf(e)) })
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
go func() {
|
||||
errs <- follow(ctx, tw, func(e jetstream.KeyValueEntry) {
|
||||
if turns.Put(e.Key(), e.Value(), !opOf(e)) {
|
||||
// Said, so a wait on a held check can be read (and a test waits on it rather than on a pause).
|
||||
logger.Printf("the gate's merge turn %s changed (revision %d): the checks it holds are held or claimed from now on", e.Key(), e.Revision())
|
||||
ch.Reconsider(ctx) // a turn went or changed: a check it held is claimed now
|
||||
}
|
||||
})
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
lazy.ensure()
|
||||
afterFirstViews := func() error {
|
||||
// The modules say which repository holds a TypeScript module, so the Node build module judges it too.
|
||||
pw, err := settle(ctx, packages, func(e jetstream.KeyValueEntry) { published.Put(e.Key(), e.Value(), !opOf(e)) })
|
||||
|
||||
@@ -7,9 +7,7 @@ import (
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
|
||||
)
|
||||
|
||||
@@ -17,26 +15,6 @@ import (
|
||||
// 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 := meshdeliveryclient.EncodeTurn(meshdelivery.TurnsByRepositoryValue{Repository: repo, Waiting: []meshdelivery.TurnsByRepositoryValueWaitingItem{{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 := meshdeliveryclient.TurnsKey(repo)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return k
|
||||
}
|
||||
|
||||
// waitFor polls a condition the checker's goroutines make true, failing after ten seconds: a bound a loaded machine does
|
||||
// not reach, waited out only by a failing test.
|
||||
func waitFor(t testing.TB, what string, ok func() bool) {
|
||||
@@ -59,66 +37,15 @@ func pullWithFP(repo string, number int, head string, d string) meshcatalog.Cand
|
||||
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) {
|
||||
// Design 47 G §278 (test 52): a new check is claimed at once; the checker reads no state of the gate, so no merge turn holds it.
|
||||
"a new check is claimed at once": 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 != meshdeliverycheckerclient.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, _ := meshcatalog.DecodeCandidatesByGroupValue(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 != meshdeliverycheckerclient.VerdictPass {
|
||||
t.Errorf("judged %d times; the running check is not ended by the turn", j.count())
|
||||
t.Errorf("judged %d times, want once at once", j.count())
|
||||
}
|
||||
},
|
||||
// §1, §5: a trunk move during a running check calls the judge once, writes its verdict, keeps the status pending (a pass
|
||||
@@ -280,15 +207,7 @@ func TestOneCheckPerDelivery(t *testing.T) {
|
||||
r.ch.Wait()
|
||||
a, _ := meshcatalog.DecodeCandidatesByGroupValue(candA)
|
||||
c, _ := meshcatalog.DecodeCandidatesByGroupValue(candC)
|
||||
at := time.Date(2026, 10, 26, 9, 0, 0, 0, time.UTC)
|
||||
turn, err := meshdeliveryclient.EncodeTurn(meshdelivery.TurnsByRepositoryValue{Repository: repoA, Waiting: []meshdelivery.TurnsByRepositoryValueWaitingItem{
|
||||
{Group: "feat/a", Heads: GateHeads(a), Approved: at, Since: at}, {Group: "feat/c", Heads: GateHeads(c), Approved: at.Add(time.Minute), Since: at}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
turns := &Turns{}
|
||||
turns.Put(turnKey(t, repoA), turn, false)
|
||||
r.ch.Turns = turns
|
||||
_, _ = a, c
|
||||
r.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS}) // a third merge moves the trunk under both
|
||||
r.ch.TrunkMoved(ctx)
|
||||
r.ch.Wait()
|
||||
|
||||
@@ -16,7 +16,6 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
|
||||
@@ -86,11 +85,6 @@ type Checker struct {
|
||||
// Runs says which commit the mesh runs of a module (the indexer's modules): the node-engine a verdict read beside at its
|
||||
// running commit is out of date when that commit moves (ADR 0548 §6); nil reads every sibling at its trunk.
|
||||
Runs interface{ CommitOf(module 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, meshdelivery.TurnsByRepositoryValueWaitingItem, bool)
|
||||
}
|
||||
// Bound is how long one check may run; none when 0 (DefaultBound).
|
||||
Bound time.Duration
|
||||
// Rebuilds hears the builder's build keys a failed verdict names (failedBuilds): once each is written done, the check of
|
||||
@@ -120,7 +114,6 @@ 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
|
||||
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
|
||||
rebuilt map[string]bool // by verdict key: every failed build its verdict names is done again
|
||||
@@ -184,7 +177,7 @@ func (s superseded) Unwrap() error { return buildrun.Withdrawn{Why: s.Error()} }
|
||||
func (ch *Checker) init() {
|
||||
ch.once.Do(func() {
|
||||
ch.known, ch.inflight, ch.again, ch.tried = map[string]meshcatalog.CandidatesByGroupValue{}, map[string]bool{}, map[string]bool{}, map[string]bool{}
|
||||
ch.nodesFail, ch.held = map[string]uint64{}, map[string]string{}
|
||||
ch.nodesFail = map[string]uint64{}
|
||||
ch.live, ch.settling = map[string]liveCheck{}, map[string]*window{}
|
||||
ch.rebuilt, ch.awaiting = map[string]bool{}, map[string]bool{}
|
||||
ch.outsideSaid = map[string]bool{}
|
||||
@@ -695,9 +688,6 @@ 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
|
||||
@@ -728,7 +718,7 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64
|
||||
ch.mu.Unlock()
|
||||
retake = !ok || epoch > at
|
||||
}
|
||||
if !retake || ch.heldByTurn(cs) {
|
||||
if !retake {
|
||||
return nil, nil
|
||||
}
|
||||
rev, err = ch.Verdicts.Update(ctx, cs.Key, data, rev)
|
||||
@@ -741,50 +731,6 @@ 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
|
||||
}
|
||||
|
||||
// ConflictWords are what the words of a head that git would not merge onto its trunk say before the files in conflict.
|
||||
const ConflictWords = "does not merge into the trunk: files in conflict: "
|
||||
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
|
||||
checker "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
|
||||
)
|
||||
|
||||
@@ -21,8 +20,7 @@ const (
|
||||
|
||||
// GateModule holds the seat mesh-delivery's states source and turns, which the checker reads (reads: mesh-delivery-gate.source,
|
||||
// mesh-delivery-gate.turns).
|
||||
GateModule = "mesh-delivery-gate"
|
||||
TurnsBucket = GateModule + "_" + meshdelivery.StateTurns
|
||||
GateModule = "mesh-delivery-gate"
|
||||
|
||||
// IndexerModule holds the seat mesh-catalog's states, which the checker follows by their views (consumes:
|
||||
// mesh-delivery-indexer.views.*).
|
||||
|
||||
@@ -51,8 +51,6 @@ 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)
|
||||
// TurnState is the gate's merge turns (ADR 0548 §3); without it no check is named as held by a turn.
|
||||
TurnState StateRead
|
||||
// Runs says which commit the mesh runs of a module (the indexer's modules), as the checker's decision reads it: the
|
||||
// node-engine a verdict names is compared at its running commit, not at the trunk's (ADR 0548 §6); nil knows none.
|
||||
Runs interface{ CommitOf(module string) string }
|
||||
@@ -411,9 +409,6 @@ 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"`
|
||||
// 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"`
|
||||
// CannotBeJudged are the candidates no pull request of which can be judged, by group, with why: the gate's source.main
|
||||
// names no trunk head of their repositories (issue 1422). They are judged once it does; until then their status is error.
|
||||
CannotBeJudged map[string]string `json:"cannot_be_judged"`
|
||||
@@ -451,42 +446,6 @@ func (r Reader) deadLetters() DeadLettersHeard {
|
||||
return DeadLettersHeard{Heard: a.Healthy, Why: a.Why}
|
||||
}
|
||||
|
||||
// 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 meshcatalog.CandidatesByGroupValue, 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.
|
||||
func (r Reader) Status() (Status, error) {
|
||||
st := Status{RunningByNode: map[string]int{}, Lost: map[string]string{}}
|
||||
@@ -503,8 +462,7 @@ func (r Reader) Status() (Status, error) {
|
||||
}
|
||||
st.Unjudged = []string{}
|
||||
st.Waiting = map[string][]string{KindNew: {}, KindNodes: {}, KindTrunk: {}}
|
||||
st.HeldByTurn, st.CannotBeJudged, st.Outside = map[string]string{}, map[string]string{}, map[string]string{}
|
||||
turns := r.turns(&st)
|
||||
st.CannotBeJudged, st.Outside = map[string]string{}, map[string]string{}
|
||||
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")
|
||||
@@ -571,9 +529,6 @@ 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 := meshdeliverycheckerclient.DecodeVerdict(raw)
|
||||
@@ -597,9 +552,6 @@ func (r Reader) Status() (Status, error) {
|
||||
case heads != nil:
|
||||
if cs := CaseOf(c, heads); cs != nil && (!sameTrunk(v.Trunk, cs.Trunk) || besideMoved(v.Beside, besideAt(heads, r.Runs))) {
|
||||
st.Waiting[KindTrunk] = append(st.Waiting[KindTrunk], c.Group)
|
||||
if why := heldBy(turns, c, heads); why != "" {
|
||||
st.HeldByTurn[c.Group] = why
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -316,18 +316,6 @@ 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, _ := meshcatalogclient.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.
|
||||
|
||||
@@ -1,79 +0,0 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
|
||||
)
|
||||
|
||||
// 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]meshdelivery.TurnsByRepositoryValue
|
||||
}
|
||||
|
||||
// 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 := meshdeliveryclient.ParseTurnsKey(key)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
if t.byRepo == nil {
|
||||
t.byRepo = map[string]meshdelivery.TurnsByRepositoryValue{}
|
||||
}
|
||||
cur, had := t.byRepo[repository]
|
||||
if !deleted {
|
||||
if turn, err := meshdeliveryclient.DecodeTurn(value); err == nil {
|
||||
t.byRepo[repository] = turn
|
||||
return !had || !meshdeliveryclient.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, meshdelivery.TurnsByRepositoryValueWaitingItem, bool) {
|
||||
if t == nil {
|
||||
return "", meshdelivery.TurnsByRepositoryValueWaitingItem{}, 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 || !meshdeliveryclient.TurnHolds(turn, heads) {
|
||||
continue
|
||||
}
|
||||
for _, w := range turn.Waiting {
|
||||
if w.Heads != heads {
|
||||
return r, w, true
|
||||
}
|
||||
}
|
||||
}
|
||||
return "", meshdelivery.TurnsByRepositoryValueWaitingItem{}, false
|
||||
}
|
||||
|
||||
// All is every turn held now, by repository (a copy).
|
||||
func (t *Turns) All() map[string]meshdelivery.TurnsByRepositoryValue {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
out := make(map[string]meshdelivery.TurnsByRepositoryValue, len(t.byRepo))
|
||||
for r, turn := range t.byRepo {
|
||||
out[r] = turn
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -112,7 +112,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-delivery-builder.builds", "mesh-delivery-gate.turns", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
"mesh-delivery-builder.builds", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
// DLV-M2 (ADR 0615 §1, ADR 0596): the indexer's candidates, modules, manifests, packages and predefined keys and the
|
||||
@@ -202,7 +202,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
|
||||
}
|
||||
// D-11 (ADR 0634 §1): the holders and buckets internal/check names, no contract naming a holder, are the manifest's: a state
|
||||
// the checker reads is a `reads` (<module>.<state>) and a state of the indexer it follows a `consumes` of its views family.
|
||||
for bucket, read := range map[string]string{check.TurnsBucket: "mesh-delivery-gate.turns",
|
||||
for bucket, read := range map[string]string{
|
||||
check.PresenceBucket: "mesh-delivery-watchman.presence", check.HaveBucket: "node-engine.have"} {
|
||||
if strings.Replace(bucket, "_", ".", 1) != read || !slices.Contains(m.Reads, read) {
|
||||
t.Errorf("the bucket %s is not the declared read %s", bucket, read)
|
||||
|
||||
@@ -53,7 +53,6 @@
|
||||
"mesh-build-node.steps",
|
||||
"mesh-build-node.toolchains",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-gate.turns",
|
||||
"mesh-delivery-watchman.presence",
|
||||
"node-engine.have"
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user