package main import ( "errors" "fmt" "os" "path/filepath" "strings" "sync" "sync/atomic" "time" ) // **The build agent says whether it is busy, and being asked drains it** (novox/hq ADR 0335 §1, §5). Its // container declares `busy` as `mesh-builder busy`: the node-engine runs it inside the container before it // replaces it. It exits 0 while a build is in hand and 1 when none is; and from the first ask the agent takes // no new build request, so its answer ends once the build in hand ends — a replaced builder finishes the check // it runs, instead of the build queue handing it out again from the start (issue 594). // // Two marks in the container's own temporary directory, as the ready mark (ready.go): a new container starts // without them, and a restarted one removes them before it takes work again. // // - in hand: written by the agent when it claims a request, before it reads whether it is drained; removed // when the build ends. // - drained: written by the ask, before it reads whether a build is in hand. // // Each side writes its own mark before it reads the other's, so an ask racing a fetch sees the build in hand, // or the agent sees the drain and hands the request back; never neither. func inHandMark() string { return filepath.Join(os.TempDir(), "mesh-builder.in-hand") } func drainMark() string { return filepath.Join(os.TempDir(), "mesh-builder.drained") } // The drain lapses when no ask has come for drainLapse: the node-engine asks every 20 seconds for as long as it // really waits, so an agent whose replacement was withdrawn — the declaration put back, the walk stopped — // takes work again rather than staying drained for ever (review of ADR 0335). And an agent drained for longer // than the node-engine's bound and the slack says it is not ready: the replacement it was drained for never // came, which is a fault to see. const ( drainLapse = 2 * time.Minute drainTooLong = 15*time.Minute + 5*time.Minute ) // drainedAt is when the drain began and when it was last asked, from its mark; false when there is no mark. func drainedAt() (began, asked time.Time, ok bool) { info, err := os.Stat(drainMark()) if err != nil { return time.Time{}, time.Time{}, false } asked = info.ModTime() began = asked if raw, err := os.ReadFile(drainMark()); err == nil { if t, err := time.Parse(time.RFC3339, strings.TrimSpace(string(raw))); err == nil && !t.After(asked) { began = t } } return began, asked, true } // drained says the agent was asked whether it is busy within drainLapse: it takes no new build request. func drained() bool { return drainedSince(time.Now()) } func drainedSince(now time.Time) bool { _, asked, ok := drainedAt() return ok && now.Sub(asked) < drainLapse } // drainedTooLong is the health check's second half: nil unless the agent has been drained, asked all along, for // longer than its node-engine's bound and the slack. func drainedTooLong(now time.Time) error { began, asked, ok := drainedAt() if !ok || now.Sub(asked) >= drainLapse || now.Sub(began) <= drainTooLong { return nil } return fmt.Errorf("drained since %s and still asked: the replacement its node-engine waits for has not come "+ "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. markFailed(err) return false } _, err = f.WriteString(time.Now().UTC().Format(time.RFC3339) + "\n") if closeErr := f.Close(); err == nil { err = closeErr } if err != nil { _ = os.Remove(inHandMark()) markFailed(err) return false } if drained() { _ = os.Remove(inHandMark()) fmt.Fprintln(os.Stderr, "asked whether busy, so this agent takes no new build: handed back") return false } inHand.Store(true) return true } // holding says what is in hand, for the ask to print: written once the request is read. func holding(what string) { _ = os.WriteFile(inHandMark(), []byte(time.Now().UTC().Format(time.RFC3339)+" "+what+"\n"), 0o644) } // 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) } } // undrained clears both marks at the start of a holder: a container started again is asked afresh. func undrained() error { for _, mark := range []string{drainMark(), inHandMark()} { if err := os.Remove(mark); err != nil && !errors.Is(err, os.ErrNotExist) { return fmt.Errorf("cannot remove %s: %w", mark, err) } } return nil } // busy is the ask: it drains the agent, then says whether a build is in hand. Nil — exit 0 — while one is. func busy() (string, error) { // The mark keeps when the drain began while it is asked within drainLapse, and is written again on every // ask: its time is the last ask. began := time.Now().UTC() if was, _, ok := drainedAt(); ok && drained() { began = was.UTC() } if err := os.WriteFile(drainMark(), []byte(began.Format(time.RFC3339)+"\n"), 0o644); err != nil { // Not drained, it could take the next build: said, and answered busy only for what is in hand now. fmt.Fprintf(os.Stderr, "cannot mark this agent drained (%s): %v\n", drainMark(), err) } held, err := os.ReadFile(inHandMark()) if err != nil { return "", errors.New("idle: no build in hand, and this agent takes no new one") } since, what, _ := strings.Cut(strings.TrimSpace(string(held)), " ") if what == "" { what = "a build" } return fmt.Sprintf("busy: %s, in hand since %s; taking no new build", what, since), nil }