Judge a duplicate hand-out before the in-hand claim, and treat a stale mark as stale (issue 1768)
mesh/merge-gate head 696300b9e280 merged into trunk f973cefd7fd3 — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 696300b9e280 merged into trunk f973cefd7fd3 — pass: its merge-check.sh passed
mesh/delivery delivered
mesh/merge-gate head 696300b9e280 merged into trunk f973cefd7fd3 — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 696300b9e280 merged into trunk f973cefd7fd3 — pass: its merge-check.sh passed
mesh/delivery delivered
A dropped duplicate left the in-hand marker, so every later claim failed and every ask spent its deliveries.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user