novox/mesh-controller
18
0
Fork 0
Code Issues Pull Requests 4 Releases Wiki Activity

Merge commit '3217f097d38d9f1c4302ec827ad6b371fd8ba95b' into feat/1696-declaration-files

Browse Source
This commit is contained in:
jschoubben
2026-10-21 18:21:37 +02:00
parent 0b11b6c98d 3217f097d3
commit 610d88e6bf
4 changed files with 218 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/link/builds_nats.go
+37
View File
@@ -362,6 +362,14 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
_ = msg.Nak()
continue
}
// **A duplicate of a check another builder is running is dropped** (issue 1649): an ask redelivered to this machine
// because its first holder did not say, in time, that it had it. The running key names that builder and this same
// ask; running it a second time would spend a second place on one check.
if why := m.runningElsewhere(ctx, request); why != "" {
fmt.Fprintf(os.Stderr, "%s is handed out again while another builder runs it and is dropped: %s\n", request.ID, why)
_ = msg.Ack()
continue
}
// A build outlives the acknowledgement window many times over; said while it runs,
// as the controller says it for its own long handlers, so the server neither hands
// the ask to a second machine nor counts the wait against its deliveries.
@@ -376,6 +384,35 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
}
}
// runningElsewhere says why an ask need not run here, or nothing: the bucket of the checks running says another builder
// runs this very ask (its id, its head and its trunk commit). A key that cannot be read runs the check.
func (m *natsMachine) runningElsewhere(ctx context.Context, r BuildRequest) string {
if r.Check == nil {
return ""
}
running, ok, err := ReadRunning(ctx, m.js.Conn(), r)
if err != nil {
fmt.Fprintf(os.Stderr, "could not read whether %s is running elsewhere, so it runs: %v\n", r.ID, err)
return ""
}
return duplicateOf(running, ok, r, m.on)
}
// runningFreshFor is how recent a running key's last heartbeat must be for its builder to be taken as alive: two missed
// heartbeats of ten seconds are not yet a dead builder, and a redelivery after a crash comes a minute after the last one.
const runningFreshFor = 25 * time.Second
// duplicateOf says why an ask is a second hand-out of one another builder is running, or nothing.
func duplicateOf(running CheckRunning, found bool, r BuildRequest, me string) string {
// Seen is the server's time and runningFreshFor is judged against this builder's clock: a skew over about 35 s would matter.
// A key not renewed for runningFreshFor is a builder that has gone: its ask is redelivered after the ack wait, and
// that is the recovery, not a duplicate (the key itself lives two minutes).
if found && running.Build == r.ID && running.Builder != me && running.Same(r) && time.Since(running.Seen) < runningFreshFor {
return fmt.Sprintf("%s is running it as %s since %s", running.Builder, running.Build, running.Started.Local().Format("15:04:05"))
}
return ""
}
// bind binds a queue's worker, which exists and is not created here.
//
// **The consumer's own filter, not the one subject this machine cares about.** The client checks what is asked for
internal/link/checks_nats.go
+18 -1
View File
@@ -253,11 +253,25 @@ func watchPullIn(ctx context.Context, buckets []string, open func(ctx context.Co
// checkTakeWait is how long a taking loop waits on one queue for an ask that is not there, and takeIdleWait how long it
// waits on the first when all three are empty.
//
// **A fetch's pull request must end before the fetch does** (issue 1649). nats.go gives the server-side request an expiry a tenth
// shorter than the fetch's wait (fetchExpiryOf), and the server then answers with a status message at the expiry; a wait so
// short that the answer can land after the fetch gave up left a status message in the pull's buffer, and anything a request
// could still deliver would be held by a machine that is busy. Each wait leaves at least minFetchMargin between the two.
const (
checkTakeWait = 30 * time.Millisecond
checkTakeWait = 100 * time.Millisecond
takeIdleWait = time.Second
)
// minFetchMargin is the least time between a fetch's server-side expiry and the end of its wait.
const minFetchMargin = 10 * time.Millisecond
// fetchExpiryOf is the expiry nats.go gives the pull request of a fetch that waits this long: the wait less a tenth of it, or
// less five seconds.
func fetchExpiryOf(wait time.Duration) time.Duration {
return wait - min(wait/10, 5*time.Second)
}
// CheckRunning is what a builder says of the check it is running, in the bucket of the checks running (novox/hq ADR 0406
// §3): the head and trunk commit it checks, the builder and its build. The key is the pull request's (RunningKeyOf).
type CheckRunning struct {
@@ -268,6 +282,8 @@ type CheckRunning struct {
// Started is when the check started, kept on every heartbeat: the entry's own time is the last heartbeat's, and a
// reader that asks how long a check has run reads this (novox/hq ADR 0433 §6).
Started time.Time `json:"started"`
// Seen is when the key was last put, which is the last heartbeat: read from the entry, never written (issue 1649).
Seen time.Time `json:"-"`
}
// RunningKeyOf is the key a check is said running under: its pull request's tokens, or `group.<name>` for a group's
@@ -319,6 +335,7 @@ func ReadRunning(ctx context.Context, conn *nats.Conn, r BuildRequest) (CheckRun
if err := json.Unmarshal(e.Value(), &running); err != nil {
return CheckRunning{}, false, nil // a key of a builder from before this means no match
}
running.Seen = e.Created()
return running, true, nil
}
internal/link/checks_nats_integration_test.go
+120
View File
@@ -6,6 +6,7 @@ import (
"context"
"encoding/json"
"reflect"
"strings"
"testing"
"time"
@@ -510,3 +511,122 @@ func TestAnUnacknowledgedCheckGoesToASecondBuilder(t *testing.T) {
stop()
}
}
// An ask handed out a second time to another builder while the first still runs it is acknowledged and not run: the
// running key names the other builder and this same ask (issue 1649).
func TestAnAskHandedOutAgainWhileAnotherBuilderRunsItIsNotRunTwice(t *testing.T) {
js := aBusWithTheBuildRole(t)
aKeyBucket(t, js)
aRunningBucket(t, js, 2*time.Minute)
ctx, stop := context.WithCancel(t.Context())
defer stop()
asks := BuildsOn(js, TheBuildMachine)
running := make(chan string, 4)
release := make(chan struct{})
a := MachineOverNATS(js, "builder-a")
defer a.Close()
go func() {
_ = a.Take(ctx, func(ctx context.Context, work Build) {
running <- work.Request().ID
<-release
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Failed: "no"})
_ = work.Done()
})
}()
check := aPullCheck("build-twice", 7, "head-7-000000000", "trunk-1-000000", false)
if err := asks.Ask(ctx, check); err != nil {
t.Fatal(err)
}
select {
case <-running:
case <-time.After(10 * time.Second):
t.Fatal("builder a took nothing")
}
// The bus's redelivery, as it reaches the other machine: the same ask, the same id.
if err := asks.Ask(ctx, check); err != nil {
t.Fatal(err)
}
b := MachineOverNATS(js, "builder-b")
defer b.Close()
took := make(chan string, 4)
go func() {
_ = b.Take(ctx, func(ctx context.Context, work Build) {
took <- work.Request().ID
_ = work.Done()
})
}()
select {
case id := <-took:
t.Fatalf("builder b ran %s, which builder a is running", id)
case <-time.After(3 * time.Second):
}
if n := len(queueOf(t, js).Of(AskWaiting)); n != 0 {
t.Errorf("%d request(s) still wait; builder b was to acknowledge the duplicate", n)
}
close(release)
}
// A busy machine fetches nothing, so no ask waiting in the queue is taken by it, whatever status message sits in its pulls
// (issue 1649). A pull request of our own that expires at once leaves such a message in the other checks' pull.
func TestABusyMachineTakesNoWaitingAskWhateverItsPullsHold(t *testing.T) {
js := aBusWithTheBuildRole(t)
aKeyBucket(t, js)
aRunningBucket(t, js, 2*time.Minute)
ctx, stop := context.WithCancel(t.Context())
defer stop()
asks := BuildsOn(js, TheBuildMachine)
busy := make(chan string, 4)
release := make(chan struct{})
a := MachineOverNATS(js, "builder-a").(*natsMachine)
defer a.Close()
go func() {
_ = a.Take(ctx, func(ctx context.Context, work Build) {
busy <- work.Request().ID
<-release
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Failed: "no"})
_ = work.Done()
})
}()
if err := asks.Ask(ctx, BuildRequest{ID: "busy", Repository: "/d"}); err != nil {
t.Fatal(err)
}
select {
case <-busy:
case <-time.After(10 * time.Second):
t.Fatal("builder a took nothing")
}
var other *takeQueue
for _, q := range a.subs {
if q.kind == queueOther && q.bound {
other = q
}
}
if other == nil {
t.Fatal("builder a has no pull on the other checks")
}
// The queue is empty, so this request is answered at its expiry with a status message that lands in the pull.
reply := strings.TrimSuffix(other.sub.Subject, "*") + "late"
if err := js.Conn().PublishRequest("$JS.API.CONSUMER.MSG.NEXT."+other.worker.Stream+"."+other.worker.Name, reply,
[]byte(`{"batch":1,"expires":1000000}`)); err != nil {
t.Fatal(err)
}
time.Sleep(500 * time.Millisecond)
for i, id := range []string{"wait-1", "wait-2"} {
if err := asks.Ask(ctx, aPullCheck(id, 20+i, "head-x-000000000", "trunk-1-000000", false)); err != nil {
t.Fatal(err)
}
}
time.Sleep(5 * time.Second)
info, err := js.Context().ConsumerInfo(other.worker.Stream, other.worker.Name)
if err != nil {
t.Fatal(err)
}
if info.NumAckPending != 0 || info.NumRedelivered != 0 || info.Delivered.Consumer != 0 {
t.Fatalf("a busy machine took from the queue: %d delivered, %d pending, %d redelivered", info.Delivered.Consumer,
info.NumAckPending, info.NumRedelivered)
}
if n := len(queueOf(t, js).Of(AskWaiting)); n != 2 {
t.Fatalf("%d asks wait; want the 2 nobody took", n)
}
close(release)
}
internal/link/held_asks_test.go
+43
View File
@@ -0,0 +1,43 @@
package link
import (
"testing"
"time"
)
// An ask a machine holds but does not start gets no heartbeat, and the bus hands it to another machine after the ack
// wait, so two machines run one check (issue 1649).
// A fetch's server-side pull request ends clearly before the fetch does, so the server's answer and any message it could still
// deliver come inside the fetch and never into a buffer nobody reads while the machine builds.
func TestAFetchsPullRequestEndsClearlyBeforeTheFetchDoes(t *testing.T) {
for name, wait := range map[string]time.Duration{"a check queue": checkTakeWait, "the idle wait": takeIdleWait} {
if margin := wait - fetchExpiryOf(wait); margin < minFetchMargin {
t.Errorf("%s: the pull request ends %v before a fetch of %v does; want at least %v", name, margin, wait, minFetchMargin)
}
}
}
func TestAnAskAnotherBuilderIsRunningIsADuplicateAndYoursIsNot(t *testing.T) {
r := aPullCheck("build-1", 423, "head-4771ee53aa", "trunk-0123456789", false)
run := CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1", Started: time.Now().Add(-time.Minute), Seen: time.Now().Add(-4 * time.Second)}
dead := run
dead.Seen = time.Now().Add(-60 * time.Second)
for name, c := range map[string]struct {
running CheckRunning
found bool
me string
dup bool
}{
"another builder runs this very ask": {run, true, "ace", true},
"its builder stopped renewing the key a minute ago": {dead, true, "ace", false},
"this builder runs it itself": {run, true, "shanks", false},
"nobody runs it": {CheckRunning{}, false, "ace", false},
"another ask of the same head": {CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-0"}, true, "ace", false},
"the same ask at another head": {CheckRunning{Head: "head-other-0000", Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1"}, true, "ace", false},
} {
if got := duplicateOf(c.running, c.found, r, c.me) != ""; got != c.dup {
t.Errorf("%s: duplicate = %v; want %v", name, got, c.dup)
}
}
}