Merge pull request 'Judge a duplicate hand-out before marking a build in hand, and never hand every build back over a stale mark (issue 1768)' (#429) from fix/1768-in-hand-marker into main

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)
 	}

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()
+}

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,
 	})

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)
+	}
+}

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
+		if !m.admit(ctx, msg, request) {
-		// 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()
 			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 {