From 696300b9e280387dabaa46dd8d612dd0c65dff70 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 20:57:12 +0200 Subject: [PATCH] Judge a duplicate hand-out before the in-hand claim, and treat a stale mark as stale (issue 1768) A dropped duplicate left the in-hand marker, so every later claim failed and every ask spent its deliveries. --- cmd/mesh-builder/busy.go | 71 ++++++++++++++++++++++++++++++++++- cmd/mesh-builder/busy_test.go | 61 ++++++++++++++++++++++++++++++ cmd/mesh-builder/main.go | 2 +- internal/link/admit_test.go | 28 ++++++++++++++ internal/link/builds_nats.go | 42 +++++++++++++++------ 5 files changed, 189 insertions(+), 15 deletions(-) create mode 100644 internal/link/admit_test.go diff --git a/cmd/mesh-builder/busy.go b/cmd/mesh-builder/busy.go index 0e8fe73d..b5db4f37 100644 --- a/cmd/mesh-builder/busy.go +++ b/cmd/mesh-builder/busy.go @@ -6,6 +6,8 @@ import ( "os" "path/filepath" "strings" + "sync" + "sync/atomic" "time" ) @@ -73,14 +75,77 @@ func drainedTooLong(now time.Time) error { "within %s (novox/hq ADR 0335)", began.UTC().Format(time.RFC3339), drainTooLong) } +// inHand is true while this process has claimed a build and not yet released it. The mark in the temporary directory +// is only believed while it is: a mark with no build of this process behind it is stale (a crash, a fault in an +// earlier path) and is removed, because left alone it hands back every ask for ever (issue 1768). +var inHand atomic.Bool + +// markRetry is how long an agent that cannot mark a build in hand takes no work, and markFault when it last failed to. +// Taking nothing is the point: an ask fetched and handed back at once, many times a second, spends all its deliveries. +const markRetry = 30 * time.Second + +var ( + markMu sync.Mutex + markFault time.Time + markSaid string +) + +// markFaulty says the agent could not mark a build in hand and has not yet been able to since: it takes no work. Once +// markRetry has passed it tries the mark itself, with no ask in hand, so that an ask is fetched only when marking works. +func markFaulty(now time.Time) bool { + markMu.Lock() + failed := markFault + markMu.Unlock() + if failed.IsZero() { + return false + } + if now.Sub(failed) < markRetry { + return true + } + f, err := os.OpenFile(inHandMark(), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644) + if err != nil && errors.Is(err, os.ErrExist) && !inHand.Load() { + _ = os.Remove(inHandMark()) + f, err = os.OpenFile(inHandMark(), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644) + } + if err != nil { + markFailed(err) + return true + } + _ = f.Close() + _ = os.Remove(inHandMark()) + markMu.Lock() + markFault, markSaid = time.Time{}, "" + markMu.Unlock() + fmt.Fprintf(os.Stderr, "the mark of a build in hand can be made again (%s): taking builds\n", inHandMark()) + return false +} + +// markFailed notes that a build could not be marked in hand, and says it when it is a new reason. +func markFailed(err error) { + markMu.Lock() + defer markMu.Unlock() + markFault = time.Now() + if said := err.Error(); said != markSaid { + markSaid = said + fmt.Fprintf(os.Stderr, "cannot mark a build in hand (%s), so this agent takes no build for %s and tries again: %v\n", + inHandMark(), markRetry, err) + } +} + // claim marks a request in hand, unless the agent is drained: then nothing is in hand, and the request goes // back to the queue. func claim() bool { + if !inHand.Load() { + // Nothing of this process is in hand, so a mark already there is stale. + if err := os.Remove(inHandMark()); err == nil { + fmt.Fprintf(os.Stderr, "removed the mark of a build in hand (%s): no build of this agent is in hand\n", inHandMark()) + } + } // Made, never written over: one build at a time, and a mark already there is a build in hand. f, err := os.OpenFile(inHandMark(), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644) if err != nil { // Unmarked, an ask would say idle while this build runs: the build is not taken. - fmt.Fprintf(os.Stderr, "cannot mark a build in hand (%s), so it is handed back: %v\n", inHandMark(), err) + markFailed(err) return false } _, err = f.WriteString(time.Now().UTC().Format(time.RFC3339) + "\n") @@ -89,7 +154,7 @@ func claim() bool { } if err != nil { _ = os.Remove(inHandMark()) - fmt.Fprintf(os.Stderr, "cannot mark a build in hand (%s), so it is handed back: %v\n", inHandMark(), err) + markFailed(err) return false } if drained() { @@ -97,6 +162,7 @@ func claim() bool { fmt.Fprintln(os.Stderr, "asked whether busy, so this agent takes no new build: handed back") return false } + inHand.Store(true) return true } @@ -107,6 +173,7 @@ func holding(what string) { // released ends the build in hand. func released() { + inHand.Store(false) if err := os.Remove(inHandMark()); err != nil && !errors.Is(err, os.ErrNotExist) { fmt.Fprintf(os.Stderr, "cannot remove the mark of the build in hand (%s): %v\n", inHandMark(), err) } diff --git a/cmd/mesh-builder/busy_test.go b/cmd/mesh-builder/busy_test.go index 1d02a748..e44da2a4 100644 --- a/cmd/mesh-builder/busy_test.go +++ b/cmd/mesh-builder/busy_test.go @@ -160,3 +160,64 @@ func TestAnUnknownCommandFails(t *testing.T) { t.Fatalf("help failed: %v", err) } } + +// 1768/AC2: a mark naming no build of this agent (left by a crash or a fault) is stale: the next ask removes it and is +// run, never fetched and handed back in a loop. +func TestAStaleInHandMarkIsRemovedAndTheAskRuns(t *testing.T) { + t.Setenv("TMPDIR", t.TempDir()) + inHand.Store(false) + if err := undrained(); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(inHandMark(), []byte("2026-10-21T18:45:06Z\n"), 0o644); err != nil { + t.Fatal(err) + } + if !claim() { + t.Fatal("a stale in-hand mark made the agent hand back an ask") + } + // A live one — this agent's own build in hand — is never taken over. + if claim() { + t.Fatal("a second build was taken while one is in hand") + } + if _, err := busy(); err != nil { + t.Fatalf("the build in hand was lost: %v", err) + } + released() +} + +// 1768/AC3: an agent that cannot mark a build in hand for another reason takes no work, instead of fetching asks and +// handing them back; it takes work again once marking works. +func TestAnAgentThatCannotMarkABuildTakesNoWorkUntilItCan(t *testing.T) { + dir := t.TempDir() + t.Setenv("TMPDIR", dir) + inHand.Store(false) + markFault, markSaid = time.Time{}, "" + t.Cleanup(func() { markFault, markSaid = time.Time{}, "" }) + if err := undrained(); err != nil { + t.Fatal(err) + } + // The mark's place is a directory that cannot be removed or made over: a directory named as the mark. + if err := os.MkdirAll(inHandMark()+"/x", 0o755); err != nil { + t.Fatal(err) + } + if claim() { + t.Fatal("an agent that cannot mark a build took it") + } + if !markFaulty(time.Now()) { + t.Fatal("an agent that cannot mark a build still takes work") + } + // Still failing after the retry time: still takes nothing, and no ask was needed to find out. + if !markFaulty(time.Now().Add(2 * markRetry)) { + t.Fatal("an agent whose mark still cannot be made takes work") + } + if err := os.RemoveAll(inHandMark()); err != nil { + t.Fatal(err) + } + if markFaulty(time.Now().Add(4 * markRetry)) { + t.Fatal("an agent whose mark can be made again takes no work") + } + if !claim() { + t.Fatal("an agent that can mark takes no build") + } + released() +} diff --git a/cmd/mesh-builder/main.go b/cmd/mesh-builder/main.go index db7cd778..3ac93d61 100644 --- a/cmd/mesh-builder/main.go +++ b/cmd/mesh-builder/main.go @@ -184,7 +184,7 @@ func run() error { // Paused or drained, it takes nothing new (novox/hq ADR 0219, ADR 0335): a drained agent finishes the build // in hand and then says idle, and its container is replaced between builds. machine := link.MachineOverNATSWith(js, on, seat, link.MachineOptions{ - Paused: func() bool { return h.Paused() || drained() }, + Paused: func() bool { return h.Paused() || drained() || markFaulty(time.Now()) }, Taking: func() { taking(seat) }, Claim: claim, }) diff --git a/internal/link/admit_test.go b/internal/link/admit_test.go new file mode 100644 index 00000000..f73b9812 --- /dev/null +++ b/internal/link/admit_test.go @@ -0,0 +1,28 @@ +package link + +import ( + "context" + "testing" + + "github.com/nats-io/nats.go" +) + +// 1768/AC1: a request dropped as a duplicate of one another builder runs marks nothing in hand: the claim is not made, +// so no mark is left behind to fail every later claim. +func TestADuplicateDropIsJudgedBeforeTheClaimSoLeavesNoMark(t *testing.T) { + claims := 0 + m := &natsMachine{ + opts: MachineOptions{Claim: func() bool { claims++; return true }}, + duplicate: func(context.Context, BuildRequest) string { return "ace is running it" }, + } + if m.admit(context.Background(), &nats.Msg{}, BuildRequest{ID: "build-1"}) { + t.Fatal("a duplicate was admitted") + } + if claims != 0 { + t.Fatalf("a dropped duplicate was claimed %d times, which leaves the in-hand mark behind", claims) + } + m.duplicate = func(context.Context, BuildRequest) string { return "" } + if !m.admit(context.Background(), &nats.Msg{}, BuildRequest{ID: "build-2"}) || claims != 1 { + t.Fatalf("a request that is no duplicate was not claimed once: claims=%d", claims) + } +} diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index b901e9fe..cf2c8402 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -162,6 +162,8 @@ type natsMachine struct { // build waits, then the checks of pull requests marked ready, then every other check (novox/hq ADR 0406 §3). subs []*takeQueue opts MachineOptions + // duplicate stands in for runningElsewhere in tests; nil is the real one. + duplicate func(context.Context, BuildRequest) string } // takeQueue is one queue a machine pulls from. @@ -356,18 +358,7 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) continue } } - // **Drained in the moment it was fetched** (novox/hq ADR 0335): handed back at once, as a pause - // landing then is. - if m.opts.Claim != nil && !m.opts.Claim() { - _ = 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() + if !m.admit(ctx, msg, request) { continue } // A build outlives the acknowledgement window many times over; said while it runs, @@ -384,6 +375,33 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) } } +// admit decides whether a fetched request is run here, and answers the message when it is not. +// +// **The duplicate is judged before the request is claimed** (issue 1768): a claim marks the build in hand, and a request +// dropped after it left the mark behind, so every later claim failed and every ask was handed back at once until all +// had spent their deliveries. A request dropped here has marked nothing. +// +// **Drained in the moment it was fetched** (novox/hq ADR 0335): handed back at once, as a pause landing then is. +func (m *natsMachine) admit(ctx context.Context, msg *nats.Msg, request BuildRequest) bool { + // **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. + elsewhere := m.runningElsewhere + if m.duplicate != nil { + elsewhere = m.duplicate + } + if why := elsewhere(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() + return false + } + if m.opts.Claim != nil && !m.opts.Claim() { + _ = msg.Nak() + return false + } + return true +} + // 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 {