Let a drain lapse, bound the gate's wait on the controller's clock, keep a busy wait out of apply time
mesh/delivery-group group feat/335-a-busy-container-is-replaced-when-idle checking: 1 of 2 member(s) ready
mesh/delivery superseded: a newer head of the same pull request
mesh/merge-gate head b0753e2aac95 merged into trunk 72214ae989b3 — pass: builds build-agent, mesh-controller, route-proxy → ace, g14, novox, shanks; no bus…
mesh/repo-check head b0753e2aac95 merged into trunk 72214ae989b3 — pass: its merge-check.sh passed

Review of #283: a drain whose replacement was withdrawn held the agent for
ever with its health green; the gate trusted the engine's until; a machine
of the rest waiting at the bound read as silent and its apply time held the
wait; an unknown mesh-builder word exited 0; plan said nothing of a busy an
older engine is not sent; and a node with a report and no send broke the
reading of every last report (the store test's NULL).
This commit is contained in:
2026-10-14 00:07:31 +02:00
parent a69c0e50dc
commit b0753e2aac
12 changed files with 403 additions and 35 deletions
+50 -5
View File
@@ -28,10 +28,49 @@ import (
func inHandMark() string { return filepath.Join(os.TempDir(), "mesh-builder.in-hand") }
func drainMark() string { return filepath.Join(os.TempDir(), "mesh-builder.drained") }
// drained says the agent was asked whether it is busy: it takes no new build request.
func drained() bool {
_, err := os.Stat(drainMark())
return err == nil
// 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)
}
// claim marks a request in hand, unless the agent is drained: then nothing is in hand, and the request goes
@@ -85,7 +124,13 @@ func undrained() error {
// 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) {
if err := os.WriteFile(drainMark(), []byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644); err != nil {
// 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)
}
+85
View File
@@ -1,9 +1,11 @@
package main
import (
"os"
"strings"
"sync"
"testing"
"time"
)
// `mesh-builder busy` answers 0 while a build is in hand and non-zero when idle, and after its first ask the
@@ -75,3 +77,86 @@ func TestAnAskRacingAFetchNeverMissesTheBuild(t *testing.T) {
released()
}
}
// A drain no longer asked lapses: an agent whose replacement was withdrawn after the first ask (a put-back, a
// revert, a stopped walk) takes work again after drainLapse, rather than staying drained with its health green.
func TestADrainNoLongerAskedLapses(t *testing.T) {
t.Setenv("TMPDIR", t.TempDir())
if err := undrained(); err != nil {
t.Fatal(err)
}
if _, err := busy(); err == nil {
t.Fatal("an idle agent answered busy")
}
if !drained() || claim() {
t.Fatal("an agent just asked takes work")
}
// The last ask was three minutes ago.
old := time.Now().Add(-3 * time.Minute)
if err := os.Chtimes(drainMark(), old, old); err != nil {
t.Fatal(err)
}
if drained() {
t.Fatal("a drain no longer asked for three minutes still holds")
}
if !claim() {
t.Fatal("an agent whose drain lapsed takes no work")
}
released()
}
// An agent drained, and asked all along, for longer than its node-engine's bound and the slack is not ready: the
// replacement it was drained for never came. One asked again after a lapse starts a new drain.
func TestAnAgentDrainedTooLongIsNotReady(t *testing.T) {
t.Setenv("TMPDIR", t.TempDir())
if err := undrained(); err != nil {
t.Fatal(err)
}
taking("mesh-build-machine")
if _, err := busy(); err == nil {
t.Fatal("an idle agent answered busy")
}
now := time.Now()
if err := drainedTooLong(now); err != nil {
t.Fatalf("an agent drained a moment ago is not ready: %v", err)
}
// The drain began 25 minutes ago, and the node-engine asked a moment ago.
began := now.Add(-25 * time.Minute).UTC().Format(time.RFC3339)
if err := os.WriteFile(drainMark(), []byte(began+"\n"), 0o644); err != nil {
t.Fatal(err)
}
err := drainedTooLong(time.Now())
if err == nil || !strings.Contains(err.Error(), "has not come") {
t.Fatalf("an agent drained for 25 minutes reads ready: %v", err)
}
// Each ask keeps when the drain began.
if _, err := busy(); err == nil {
t.Fatal("an idle agent answered busy")
}
if err := drainedTooLong(time.Now()); err == nil {
t.Fatal("an ask started the drain again")
}
// The asks stopped: the drain lapsed, and so did the fault.
old := time.Now().Add(-3 * time.Minute)
if err := os.Chtimes(drainMark(), old, old); err != nil {
t.Fatal(err)
}
if err := drainedTooLong(time.Now()); err != nil {
t.Fatalf("a lapsed drain still reads not ready: %v", err)
}
}
// An ask this build does not know exits non-zero: an older image asked `busy` reads idle, not busy for the
// node-engine's whole bound.
func TestAnUnknownCommandFails(t *testing.T) {
was := os.Args
t.Cleanup(func() { os.Args = was })
os.Args = []string{"mesh-builder", "some-later-word"}
if err := run(); err == nil {
t.Fatal("an unknown command exited 0")
}
os.Args = []string{"mesh-builder", "help"}
if err := run(); err != nil {
t.Fatalf("help failed: %v", err)
}
}
+12 -3
View File
@@ -92,8 +92,12 @@ func run() error {
case "build":
return buildOnce(context.Background(), os.Args[2:])
case "ready":
// The container's health check (novox/hq issue 544): exits non-zero until the agent takes work.
return ready()
// The container's health check (novox/hq issue 544): exits non-zero until the agent takes work, and
// once it has been drained longer than its node-engine would wait to replace it (ADR 0335).
if err := ready(); err != nil {
return err
}
return drainedTooLong(time.Now())
case "busy":
// The container's busy command (novox/hq ADR 0335): drains the agent, and exits non-zero once
// nothing is in hand.
@@ -103,9 +107,14 @@ func run() error {
}
fmt.Println(said)
return nil
default:
case "help", "-h", "--help":
fmt.Print(usage)
return nil
default:
// **An ask this build does not know fails** (novox/hq ADR 0335): an image older than `busy`, asked
// it, must read as idle — exiting 0 would hold its container for the node-engine's whole bound.
fmt.Fprint(os.Stderr, usage)
return fmt.Errorf("%q is not a command this mesh-builder knows", os.Args[1])
}
}
+23 -4
View File
@@ -21,6 +21,21 @@ import (
// container replaced: a pass of the engine (its asks every 20 seconds) and the report's way here.
var busyGateSlack = 5 * time.Minute
// busyEngineBound is the node-engine's bound on a wait (novox/hq ADR 0335 §3), counted again here on this
// controller's clock from when it first read the wait.
const busyEngineBound = 15 * time.Minute
// busyGateEnds is when the gate stops waiting: the engine's own word on its bound, or this controller's count of
// it from when it first read the wait, whichever is sooner, and the slack — so neither a clock ahead on the
// machine nor a wrong word from it holds a walk longer.
func busyGateEnds(b *inventory.GateBusy) time.Time {
until := b.Seen.Add(busyEngineBound)
if !b.Until.IsZero() && b.Until.Before(until) {
until = b.Until
}
return until.Add(busyGateSlack)
}
// busyWait decides, before a judging, whether the gate waits for a busy container on a machine it judges:
// true while it waits, the gate saying so. False when nothing waits, or no longer: a wait that ended moves the
// gate's Since to its end. Pure but for the gate it is handed.
@@ -34,15 +49,19 @@ func busyWait(g *inventory.PlanGate, pairs []judged, f gateFacts, now time.Time)
continue
}
if g.Busy == nil {
g.Busy = &inventory.GateBusy{Machine: j.node, Module: j.module, Container: b.Container, Since: b.Since}
g.Busy = &inventory.GateBusy{Machine: j.node, Module: j.module, Container: b.Container, Since: b.Since,
Seen: now}
}
if g.Busy.Seen.IsZero() {
g.Busy.Seen = now
}
g.Busy.Until, g.Busy.Said = b.Until, b.Said
if now.After(g.Busy.Until.Add(busyGateSlack)) {
if now.After(busyGateEnds(g.Busy)) {
// The engine said it would have replaced it by now: the judging starts, and finds the module not
// yet healthy while the machine still lists it waiting (busyStill), until the gate's own bound.
end := now
g.Busy.Ended, g.Busy.PastBound = &end, true
g.Since, g.Passes, g.LastPass, g.Healthy, g.HealthyAt = &end, 0, nil, nil, nil
g.Since, g.Passes, g.LastPass, g.Healthy, g.HealthyAt, g.Applied = &end, 0, nil, nil, nil, nil
return false
}
g.Passes, g.LastPass, g.Last, g.Failing = 0, nil, g.Busy.Words(), nil
@@ -58,7 +77,7 @@ func busyWait(g *inventory.PlanGate, pairs []judged, f gateFacts, now time.Time)
end = r.At.UTC()
}
g.Busy.Ended = &end
g.Since, g.Passes, g.LastPass, g.Last, g.Healthy, g.HealthyAt = &end, 0, nil, "", nil, nil
g.Since, g.Passes, g.LastPass, g.Last, g.Healthy, g.HealthyAt, g.Applied = &end, 0, nil, "", nil, nil, nil
return false
}
+37
View File
@@ -5,6 +5,7 @@ import (
"testing"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
@@ -109,3 +110,39 @@ func TestAGateOfAnEngineThatNeverWaitsStartsAsBefore(t *testing.T) {
t.Fatal("a wait reported about an earlier declaration held the gate")
}
}
// The wait is bounded on this controller's clock: an engine whose clock runs an hour ahead, saying it waits until
// then, holds the gate no longer than 15 minutes and the slack from when the controller first read the wait.
func TestAGateWaitIsBoundedOnTheControllersClock(t *testing.T) {
seen := time.Date(2026, 10, 13, 1, 0, 0, 0, time.UTC)
g := &inventory.PlanGate{Machines: []string{"novox"}, Since: &seen}
pairs := []judged{{module: "build-agent", node: "novox"}}
ahead := seen.Add(time.Hour)
f := gateFacts{reports: map[string]inventory.Reported{"novox": {Node: "novox", Outcome: inventory.OutcomeApplied,
At: &seen, Current: true, Busy: []inventory.BusyWait{{ID: "build-agent.server", Container: "mesh-build-agent",
State: inventory.BusyWaiting, Since: ahead, Until: ahead.Add(15 * time.Minute)}}}}}
if !busyWait(g, pairs, f, seen) || !g.Busy.Seen.Equal(seen) {
t.Fatalf("the gate did not start waiting, seen now: %+v", g.Busy)
}
if !busyWait(g, pairs, f, seen.Add(15*time.Minute+busyGateSlack-time.Second)) {
t.Fatal("the gate stopped waiting within the bound on its own clock")
}
if busyWait(g, pairs, f, seen.Add(15*time.Minute+busyGateSlack+time.Second)) || !g.Busy.PastBound {
t.Fatalf("an engine's later word held the gate past the controller's bound: %+v", g.Busy)
}
}
// `plan` says, for a machine whose node-engine is older than the field, each container whose busy is not sent.
func TestAPlanSaysABusyAnOlderEngineIsNotSent(t *testing.T) {
modules := []catalogue.Manifest{{Module: "build-agent", Resources: []map[string]any{
{"id": "server", "type": "container", "busy": map[string]any{"command": "mesh-builder busy"}},
{"id": "agent-env", "type": "file"}}}}
lines := busyNotSent(modules, false)
if len(lines) != 1 || !strings.Contains(lines[0], "build-agent.server declares busy") ||
!strings.Contains(lines[0], "replaced at once") {
t.Fatalf("the plan does not say the busy not sent: %v", lines)
}
if lines := busyNotSent(modules, true); len(lines) != 0 {
t.Fatalf("a busy sent is said as not sent: %v", lines)
}
}
+28
View File
@@ -1059,6 +1059,25 @@ func engineReadsBusy(ctx context.Context, inv *inventory.Inventory, node string)
return had && stated.Contract >= link.BusyContract, nil
}
// busyNotSent is, for a node-engine that does not read a container's busy, each container whose module declares
// one: not sent it, so replaced at once while it may hold work (novox/hq ADR 0335). Nothing for one that reads it.
func busyNotSent(modules []catalogue.Manifest, reads bool) []string {
if reads {
return nil
}
var out []string
for _, m := range modules {
for _, r := range m.Resources {
if _, has := r[catalogue.BusyField]; has {
out = append(out, fmt.Sprintf("%s.%v declares busy, and this node-engine does not read it yet (its "+
"statement's contract is below %d): not sent, so the container is replaced at once, busy or not "+
"(novox/hq ADR 0335)", m.Module, r["id"], link.BusyContract))
}
}
}
return out
}
// zonesInTheMesh is every zone a module in the mesh declares, where the mesh placed it (novox/hq ADR
// 0199): the zone settled from that node's settings, the node's private address, the port the
// answering listen is published on there.
@@ -1466,6 +1485,15 @@ func planCommand(ctx context.Context, args []string) error {
for _, line := range unboundLines(plan) {
fmt.Println(line)
}
// And a container's busy that this machine's node-engine is too old to be sent (novox/hq ADR 0335): its
// container is replaced at once, as before. A statement that cannot be read is said as such.
if reads, err := engineReadsBusy(ctx, open.inventory, args[0]); err != nil {
fmt.Printf("%s: whether its node-engine reads a container's busy could not be read: %v\n", args[0], err)
} else {
for _, line := range busyNotSent(plan.Modules, reads) {
fmt.Printf("%s: %s\n", args[0], line)
}
}
declared, err := declarationFor(ctx, open, args[0], plan, settings)
if err != nil {
return err
+55
View File
@@ -140,3 +140,58 @@ func TestAReportOfABusyContainerIsKeptAndReadBack(t *testing.T) {
t.Error("a container replaced still reads as waiting")
}
}
// A machine of the rest that answered and waits for a busy container is never read as silent, its wait is kept
// out of its apply time, and once replaced its report is the end of the wait (novox/hq ADR 0335).
func TestARestMachineWaitingForABusyContainerIsNeverSilent(t *testing.T) {
c := walk582(t)
shanks := c.walk.Modules["build-agent"].Rest["shanks"]
ref := shanks.Digest + "@" + shanks.At.Format("2006-01-02T15:04:05.000000")
answered := shanks.At.Add(15 * time.Second)
// shanks answered, and said a container waits: its apply row is the answer, marked busy.
c.rows["shanks"][0].Detail = ApplyBusy
at := c.judged.Add(20 * time.Minute) // past ApplySilentAfter from its send
w := c.walk.WalkPhases(at, c.lookup)
var got RestReport
for _, r := range w.Rest {
if r.Node == "shanks" {
got = r
}
}
if got.State != RestBusy || slicesContains(w.Silent, "shanks") {
t.Fatalf("a machine waiting for a busy container reads %q (silent: %v)", got.State, w.Silent)
}
if w.End != nil || !strings.Contains(phase(w, PhaseApply, -1).Said, "waits for a busy container on shanks") {
t.Fatalf("the walk ended, or does not say it waits: %+v", phase(w, PhaseApply, -1))
}
// 17 minutes on, shanks says it replaced the container: its wait is its own duration, and its report is then.
c.rows["shanks"] = append(c.rows["shanks"], Duration{Kind: DurationBusy, Subject: "shanks", Node: "shanks", Ref: ref,
Started: answered, Took: 17 * time.Minute})
w = c.walk.WalkPhases(c.judged.Add(time.Hour), c.lookup)
sumsUp(t, w)
for _, r := range w.Rest {
if r.Node == "shanks" {
got = r
}
}
if got.State != RestApplied || got.Busy != "17m0s" || !got.Reported.Equal(answered.Add(17*time.Minute)) {
t.Fatalf("a machine replaced after its wait reads %+v", got)
}
if slicesContains(w.Silent, "shanks") || !w.End.Equal(answered.Add(17*time.Minute)) {
t.Fatalf("the walk read shanks as silent, or ended elsewhere: %v %v", w.Silent, w.End)
}
// And the apply time itself is the answer's, never the wait's.
if r, _ := FirstReportOf(c.rows["shanks"], shanks, c.walkSent); !r.Answered.Equal(answered) || r.Busy != 17*time.Minute {
t.Fatalf("the busy wait is not kept apart from the apply: %+v", r)
}
}
func slicesContains(list []string, s string) bool {
for _, x := range list {
if x == s {
return true
}
}
return false
}
+29 -1
View File
@@ -20,10 +20,19 @@ const (
// DurationWalkPhase is one phase of an ended walk, per class (novox/hq ADR 0282 decision 6): subject
// "<class>/<phase>", and "<class>/total" for its merge to every machine running it.
DurationWalkPhase = "walk-phase"
// DurationBusy is a machine's wait for a busy container (novox/hq ADR 0335): from its first report of a
// send, which said a container waits because it is busy, to the report that says it replaced — the same
// send's ref as its `apply` row, whose detail is ApplyBusy. Its own measure, never in the apply's.
DurationBusy = "busy-wait"
)
// ApplyBusy is the detail of an `apply` duration whose report said a container waits because it is busy: the
// machine answered, and is not yet running what it was sent (novox/hq ADR 0335).
const ApplyBusy = "busy"
// DurationKinds are every kind, in the order `durations` shows them.
var DurationKinds = []string{DurationApply, DurationHeartbeatGap, DurationPlanTier, DurationBuild, DurationWalkPhase}
var DurationKinds = []string{DurationApply, DurationHeartbeatGap, DurationPlanTier, DurationBuild, DurationWalkPhase,
DurationBusy}
// DurationsKeptFor is how long a duration is kept: long enough to set a bound from, and to correct it
// in Phase 1's first live week.
@@ -70,6 +79,25 @@ func (i *Inventory) RecordApplyDuration(ctx context.Context, node, declared, out
return err
}
// RecordBusyDuration measures a machine's wait for a busy container, at the report that says it replaced
// (novox/hq ADR 0335): from the end of its `apply` duration of the send it was last sent, which said it waited, to
// now. Nothing when that send's first report did not say a wait, and once per send.
func (i *Inventory) RecordBusyDuration(ctx context.Context, node, declared string) error {
if declared == "" {
return nil
}
_, err := i.store.Pool().Exec(ctx,
`insert into duration (kind, subject, node, ref, started, took_ms, detail)
select $1, d.subject, d.node, d.ref, d.started + d.took_ms * interval '1 millisecond',
greatest(0, (extract(epoch from (now() - (d.started + d.took_ms * interval '1 millisecond'))) * 1000)::bigint), ''
from duration d join node n on n.name = d.node
where d.kind = $2 and d.subject = $3 and d.detail = $5 and n.sent = $4 and n.sent_at is not null
and d.ref = n.sent || '@' || to_char(n.sent_at at time zone 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US')
on conflict do nothing`,
DurationBusy, DurationApply, node, declared, ApplyBusy)
return err
}
// RecordHeartbeatGap measures the silence before a machine's word: from the last word heard before
// it, which the caller read before recording this one.
func (i *Inventory) RecordHeartbeatGap(ctx context.Context, node string, before time.Time) error {
+1 -1
View File
@@ -1182,7 +1182,7 @@ func (i *Inventory) SentTo(ctx context.Context, name string) (SentDeclaration, b
func (i *Inventory) LastReports(ctx context.Context) ([]Reported, error) {
rows, err := i.store.Pool().Query(ctx,
`select n.name, coalesce(r.outcome, ''), r.at, n.sent_at,
r.declared is not null and r.declared <> '' and r.declared = n.sent,
coalesce(r.declared is not null and r.declared <> '' and r.declared = n.sent, false),
coalesce(r.declared, ''), coalesce(r.reported_sequence, 0), coalesce(r.busy, '[]')
from node n left join node_report r on r.node = n.id
order by n.name`)
+5 -2
View File
@@ -401,8 +401,11 @@ type GateBusy struct {
// Since is when the machine's first apply that would have replaced it ran, Until when its node-engine
// replaces it at the latest; Ended when the machine reported it replaced and the judging began — or when
// the controller stopped waiting past the engine's bound, which PastBound says.
Since time.Time `json:"since"`
Until time.Time `json:"until"`
Since time.Time `json:"since"`
Until time.Time `json:"until"`
// Seen is when this controller first read the wait, on its own clock: the gate's bound runs from it, never
// from the engine's word alone.
Seen time.Time `json:"seen,omitzero"`
Ended *time.Time `json:"ended,omitempty"`
PastBound bool `json:"past_bound,omitempty"`
// Said is the last output of its busy command, as the machine reported it.
+56 -15
View File
@@ -91,6 +91,11 @@ type WalkPhase struct {
type AppliedReport struct {
At time.Time
Outcome string
// Answered is, for a machine that waited for a busy container (novox/hq ADR 0335), its first report of the
// send; Busy how long it then waited; Waiting that it still waits, At being its first report then.
Answered time.Time
Busy time.Duration
Waiting bool
}
// AppliedLookup answers a machine's report of the send the walk made to it: send is what the walk kept of it, and
@@ -109,9 +114,13 @@ type RestReport struct {
Reported *time.Time `json:"reported,omitempty"`
Took string `json:"took,omitempty"`
Outcome string `json:"outcome,omitempty"`
// State is applied, failed, waiting (not yet heard, within ApplySilentAfter) or silent (not heard within it:
// listed, never moving the walk's end).
// Busy is how long it waited for a busy container before it ran the build (novox/hq ADR 0335): inside Took,
// never read as the machine being silent.
Busy string `json:"busy,omitempty"`
// State is applied, failed, waiting (not yet heard, within ApplySilentAfter), busy (it answered and waits for
// a busy container: never silent) or silent (not heard within it: listed, never moving the walk's end).
State string `json:"state"`
busy time.Duration
}
// The states of a machine of the rest.
@@ -119,6 +128,7 @@ const (
RestApplied = "applied"
RestFailed = "failed"
RestWaiting = "waiting"
RestBusy = "busy"
RestSilent = "silent"
)
@@ -413,6 +423,13 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin
if r, ok := applied(s.node, s.what, s.walkSent); ok {
got := r.At
rr.Reported, rr.Outcome, rr.Took = &got, r.Outcome, words(r.At.Sub(at))
if r.Busy > 0 {
rr.Busy = words(r.Busy)
rr.busy = r.Busy
}
if r.Waiting {
rr.Reported, rr.Took, rr.State = nil, "", RestBusy
}
}
reports[key] = rr
order = append(order, key)
@@ -429,16 +446,20 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin
if later.Node == rr.Node && later.Sent.After(rr.Sent) && later.Reported != nil &&
(rr.Reported == nil || later.Reported.Before(*rr.Reported)) {
got := *later.Reported
rr.Reported, rr.Outcome, rr.Took = &got, later.Outcome, words(got.Sub(rr.Sent))
rr.Reported, rr.Outcome, rr.Took, rr.State = &got, later.Outcome, words(got.Sub(rr.Sent)), ""
}
}
}
var end *time.Time
var waiting, failed []string
var waiting, failed, busy []string
for _, k := range order {
rr := reports[k]
switch {
case rr.Reported == nil && now.Sub(rr.Sent) > ApplySilentAfter, rr.Reported != nil && rr.Reported.Sub(rr.Sent) > ApplySilentAfter:
case rr.State == RestBusy:
// It answered and waits for a busy container: never silent, and the walk's end waits for it.
busy = appendOnce(busy, rr.Node)
case rr.Reported == nil && now.Sub(rr.Sent) > ApplySilentAfter,
rr.Reported != nil && rr.Reported.Sub(rr.Sent)-rr.busy > ApplySilentAfter:
rr.State = RestSilent
w.Silent = appendOnce(w.Silent, rr.Node)
case rr.Reported == nil:
@@ -463,9 +484,15 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin
switch {
case len(failed) > 0:
return append(out, point{phase: PhaseApply, tier: -1, said: "not applied on " + strings.Join(failed, ", ")})
case len(waiting) > 0:
return append(out, point{phase: PhaseApply, tier: -1, said: "waiting for " + strings.Join(waiting, ", ") +
" to report it applied"})
case len(waiting) > 0 || len(busy) > 0:
var said []string
if len(waiting) > 0 {
said = append(said, "waiting for "+strings.Join(waiting, ", ")+" to report it applied")
}
if len(busy) > 0 {
said = append(said, "waits for a busy container on "+strings.Join(busy, ", "))
}
return append(out, point{phase: PhaseApply, tier: -1, said: strings.Join(said, "; ")})
case end == nil:
return append(out, point{phase: PhaseApply, tier: -1, said: "no machine of the rest heard from: " +
strings.Join(w.Silent, ", ")})
@@ -556,19 +583,19 @@ func words(d time.Duration) string {
func (i *Inventory) FirstAppliedAfter(ctx context.Context, node string, send SentDeclaration, walkSent time.Time) (AppliedReport, bool, error) {
from, to := applyRowsBetween(send, walkSent)
rows, err := i.store.Pool().Query(ctx,
`select ref, started, took_ms, detail from duration
where kind = $1 and subject = $2 and started >= $3 and started < $4
`select kind, ref, started, took_ms, detail from duration
where subject = $2 and ((kind = $1 and started >= $3 and started < $4) or (kind = $5 and started >= $3))
order by started`,
DurationApply, node, from, to)
DurationApply, node, from, to, DurationBusy)
if err != nil {
return AppliedReport{}, false, err
}
defer rows.Close()
var ds []Duration
for rows.Next() {
d := Duration{Kind: DurationApply, Subject: node, Node: node}
d := Duration{Subject: node, Node: node}
var ms int64
if err := rows.Scan(&d.Ref, &d.Started, &ms, &d.Detail); err != nil {
if err := rows.Scan(&d.Kind, &d.Ref, &d.Started, &ms, &d.Detail); err != nil {
return AppliedReport{}, false, err
}
d.Took = time.Duration(ms) * time.Millisecond
@@ -603,16 +630,30 @@ func applyRowsBetween(send SentDeclaration, walkSent time.Time) (time.Time, time
// before the walk's record of its send, as it was read before.
//
// A send past ApplySilentAfter is never read: its machine slept, and is listed silent (ADR 0282 decision 1).
//
// **A first report that said a container waits because it is busy** (novox/hq ADR 0335) is the machine's answer,
// and not yet the build running: its report is the end of its busy wait, the `busy-wait` duration of the same
// send, with the wait said apart (Busy). While none is kept the machine is still waiting (Waiting), never silent.
func FirstReportOf(ds []Duration, send SentDeclaration, walkSent time.Time) (AppliedReport, bool) {
from, to := applyRowsBetween(send, walkSent)
for _, d := range ds {
if d.Started.Before(from) || !d.Started.Before(to) {
if d.Kind == DurationBusy || d.Started.Before(from) || !d.Started.Before(to) {
continue
}
if send.Digest != "" && !strings.HasPrefix(d.Ref, send.Digest+"@") {
continue
}
return AppliedReport{At: d.Started.Add(d.Took).UTC(), Outcome: d.Detail}, true
answered := d.Started.Add(d.Took).UTC()
if d.Detail != ApplyBusy {
return AppliedReport{At: answered, Outcome: d.Detail}, true
}
for _, b := range ds {
if b.Kind == DurationBusy && b.Ref == d.Ref {
return AppliedReport{At: b.Started.Add(b.Took).UTC(), Outcome: OutcomeApplied, Answered: answered,
Busy: b.Took}, true
}
}
return AppliedReport{At: answered, Outcome: OutcomeApplied, Answered: answered, Waiting: true}, true
}
return AppliedReport{}, false
}
+22 -4
View File
@@ -500,13 +500,21 @@ func (e Enrolment) Heard(ctx context.Context, report Report) (news bool, err err
// And how long the machine took, from the send to this first account of it (novox/hq to-be 45
// Phase 0): what the sent-not-reported bound will be set from. Said if lost, never a failure.
//
// **Not while a container waits because it said busy** (novox/hq ADR 0335 §4): the machine is not yet
// running what it was sent, and the account that says the container replaced is the one that ends the
// apply — so a walk's rest counts the wait in its delivery time, never exempted.
// **A wait for a busy container is its own measure** (novox/hq ADR 0335): the first report that says one
// waits is the machine's answer, kept as its apply with the detail ApplyBusy; the report that says the
// container replaced ends the wait, kept as a `busy-wait` of the same send — so the apply's time holds no
// busy time, and the walk reads the machine as answered and waiting, never as silent.
outcome := doing.Outcome
if doing.Waiting() {
outcome = inventory.ApplyBusy
log.Printf("%s applied declaration %s and leaves a busy container running until it says idle: %s",
report.Node, short(report.Declared), busyWords(doing.Busy))
} else if err := e.Inventory.RecordApplyDuration(ctx, report.Node, report.Declared, doing.Outcome); err != nil {
} else if replacedAfterWaiting(doing.Busy) {
if err := e.Inventory.RecordBusyDuration(ctx, report.Node, report.Declared); err != nil {
log.Printf("how long %s waited for a busy container could not be recorded: %v", report.Node, err)
}
}
if err := e.Inventory.RecordApplyDuration(ctx, report.Node, report.Declared, outcome); err != nil {
log.Printf("how long %s took to apply could not be recorded: %v", report.Node, err)
}
@@ -530,3 +538,13 @@ func busyWords(busy []inventory.BusyWait) string {
}
return strings.Join(out, "; ")
}
// replacedAfterWaiting says a report replaced a container that had waited because it was busy.
func replacedAfterWaiting(busy []inventory.BusyWait) bool {
for _, b := range busy {
if b.State != inventory.BusyWaiting {
return true
}
}
return false
}