diff --git a/cmd/mesh-host/backup_event_test.go b/cmd/mesh-host/backup_event_test.go index 51f1bf5..fa0d55f 100644 --- a/cmd/mesh-host/backup_event_test.go +++ b/cmd/mesh-host/backup_event_test.go @@ -6,10 +6,13 @@ import ( "errors" "fmt" "strings" + "sync" "testing" "time" + "git.novox.be/novox/mesh-sdk/go/declaration" "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/apply" "github.com/novox/mesh-host/internal/link" ) @@ -85,4 +88,62 @@ func TestABackupEndsWhenItsEventSaysSo(t *testing.T) { t.Errorf("%s: asked %d times, %v", row.name, asked, err) } } + + // The events of the same kind (ENG-11): the status view's settled is told to the agent by the module that settled, on + // whichever node; a body with no module, or no JSON, tells nothing; the end of the link ends the hearing. + told := &settledRecorder{} + events := make(chan []byte, 4) + for _, body := range []string{`{"node":"n2","module":"distribution"}`, `{"node":"n1"}`, `not json`, `{"module":"gitea"}`} { + events <- []byte(body) + } + linked, end := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { defer close(done); listen(linked, events, func() settledTold { return told }, func(string) {}) }() + deadline := time.Now().Add(2 * time.Second) + for len(told.heard()) < 2 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + end() + <-done + if got := strings.Join(told.heard(), ","); got != "distribution,gitea" { + t.Errorf("the agent was told %q settled, want distribution,gitea", got) + } + // And the images an apply could not pull are said with the holder the declaration names beside each, none for a public + // one (ENG-8, ENG-10). + d, err := declaration.ParseFileTrusted([]byte(`{"declaration":1,"resources":[` + + `{"id":"a.first","type":"container","name":"first","image":"reg/x@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","image-seat":"mesh-artifact-store","image-holder":"distribution"},` + + `{"id":"a.second","type":"container","name":"second","image":"public/y@sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"}]}`)) + if err != nil { + t.Fatal(err) + } + pull := func(image string) error { + return &apply.Error{Resource: "x", Err: &apply.PullError{Image: image, Err: errors.New("manifest unknown")}} + } + if got := pullsOf(d, pull("reg/x@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")); len(got) != 1 || got[0].Seat != "mesh-artifact-store" || got[0].Holder != "distribution" { + t.Errorf("a store's image is said as %+v", got) + } + if got := pullsOf(d, pull("public/y@sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")); len(got) != 1 || got[0].Holder != "" || got[0].Seat != "" { + t.Errorf("a public image is said as %+v", got) + } + if got := pullsOf(d, errors.New("another failure")); len(got) != 0 { + t.Errorf("a failure that is no pull is said as %+v", got) + } +} + +// settledRecorder is the agent a hearing tells. +type settledRecorder struct { + mu sync.Mutex + list []string +} + +func (r *settledRecorder) Settled(holder string) { + r.mu.Lock() + r.list = append(r.list, holder) + r.mu.Unlock() +} + +func (r *settledRecorder) heard() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.list...) } diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 1f056e0..0d03f6b 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1064,6 +1064,14 @@ func runLink(ctx context.Context, opts options) error { // And what it trusts, followed on the bus, said in its record whenever it changes (ADR 0550 §2). trust.onChange(func() { factsOfNow(aside, current.get()) }) go followTrust(aside, queue, trust, say) + // And the status view's `settled`, for a failed pull from a store to be tried again once the store's holder has settled + // (design 51, ENG-11). + go followSettled(aside, queue, func() settledTold { + if ag := current.get(); ag != nil { + return ag + } + return nil + }, say) // And the login it dials as, which the vault's node-logins names for this node (novox/hq ADR 0583 §1). rotator := newRotator(queue, opts.state, opts.timeout, say) go followLogins(aside, queue, mine.Node, func(nl delivery.NodeLogin) { diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 2ee61ae..f0e67b7 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -731,12 +731,19 @@ func applyComposition(ctx context.Context, opts options, c delivery.Composed, ot if got.Refused != "" { return failedWith(got.Refused) } + // An image that could not be pulled from a store is said with its holder, for the engine to try the apply again once the + // holder has settled (design 51, ENG-8). + failedOn := func(why string) agent.Result { + res := failedWith(why) + res.Pulls = pullsOf(d, got.Err) + return res + } results := apply.ModuleResults(d, got.Outcome, got.Err) if r, ok := results[name]; ok && r.Failed { - return failedWith(r.Detail) + return failedOn(r.Detail) } if got.Err != nil { - return failedWith(got.Err.Error()) + return failedOn(got.Err.Error()) } if got.Probe != nil { return failedWith(got.Probe.partial().Error()) diff --git a/cmd/mesh-host/pulls.go b/cmd/mesh-host/pulls.go new file mode 100644 index 0000000..7703a5c --- /dev/null +++ b/cmd/mesh-host/pulls.go @@ -0,0 +1,26 @@ +package main + +import ( + "git.novox.be/novox/mesh-sdk/go/declaration" + + "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/apply" +) + +// pullsOf is the images an apply could not find or pull, each with the seat it came through and that seat's holder as the +// declaration names them beside the container (design 51 *distribution's first delivery*, ENG-8, ENG-10). An image the +// declaration names no holder for (a public registry's) carries none: its pull is not tried again on a store settling. +func pullsOf(d *declaration.Declaration, err error) []agent.PullFailure { + var out []agent.PullFailure + for _, p := range apply.PullErrors(err) { + f := agent.PullFailure{Image: p.Image, Started: p.Started} + for _, r := range d.Resources { + if c, ok := r.(*declaration.Container); ok && c.Image == p.Image && c.ImageHolder != "" { + f.Seat, f.Holder = c.ImageSeat, c.ImageHolder + break + } + } + out = append(out, f) + } + return out +} diff --git a/cmd/mesh-host/settled.go b/cmd/mesh-host/settled.go new file mode 100644 index 0000000..5a7dd82 --- /dev/null +++ b/cmd/mesh-host/settled.go @@ -0,0 +1,71 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/novox/mesh-host/internal/link" +) + +// settledEvent is where the status view says a module has settled on a node: its have reached its want, running and healthy +// (design 47; the view's `settled`, said once for a module on a node at one composition). The engine hears it for every module +// and tells its agent, which tries again, once per settled, a failed apply that waits on that module as the holder of the store +// its image comes from (design 51 *distribution's first delivery*, issue 1966 ENG-11). The grants part grants the engine this +// subject (NodeAgentHears); a refused subscription is said, and made again when the next link opens. +const settledEvent = "mesh.mod.mesh-delivery-view.event.settled" + +// settledBody is what of the event the engine reads: the module that settled, on whichever node. +type settledBody struct { + Module string `json:"module"` +} + +// settledTold is who is told a holder has settled. +type settledTold interface{ Settled(holder string) } + +// followSettled hears the status view's `settled` on every link, never on a timer, until ctx ends. now names the agent running +// now (nil while the node agent begins again): a settled heard while there is none is not lost for it, since a new agent applies +// again every have that failed at its start. +func followSettled(ctx context.Context, session *link.Session, now func() settledTold, say link.Announce) { + said := "" + for ctx.Err() == nil { + linked, cancel, err := session.WhenLinked(ctx) + if err != nil { + return + } + events, stop, herr := session.Hear(settledEvent) + if herr != nil { + if line := fmt.Sprintf("a store settling is not heard yet (%s: %v): a failed pull waits for the next link, or for compose", settledEvent, herr); line != said { + said = line + say(line) + } + } else { + said = "" + listen(linked, events, now, say) + stop() + } + <-linked.Done() + cancel() + } +} + +// listen tells the agent each settled it hears, until the link ends. +func listen(linked context.Context, events <-chan []byte, now func() settledTold, say link.Announce) { + for { + select { + case <-linked.Done(): + return + case raw, ok := <-events: + if !ok { + return + } + var b settledBody + if json.Unmarshal(raw, &b) != nil || b.Module == "" { + continue + } + if told := now(); told != nil { + told.Settled(b.Module) + } + } + } +} diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 9f63130..5a6fc1b 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -82,6 +82,9 @@ type Haves interface { // Result is how one apply or removal came out. type Result struct { + // Pulls are the images the apply could not find or pull, each with the seat it came through and that seat's holder + // where the declaration names them (a store's image, ENG-10): what a failed apply waits on (ENG-8). + Pulls []PullFailure Failed bool // Transient says the failure was the bus not answering for a moment (no responders, a timeout, a reconnect, a // JetStream API error while the bus reloads), not the module's: the apply is tried again without a new revision. @@ -89,6 +92,14 @@ type Result struct { Detail string } +// PullFailure is one image an apply could not find or pull. Holder is the module holding the seat the image came through, +// "" for an image no store holds (a public registry's): its pull is tried again only by the changes that always start an +// apply. Started is when the pull began. +type PullFailure struct { + Image, Seat, Holder string + Started time.Time +} + // Backup is the answer of the backup asked before a module moves. // BackupInterrupted is the words of a backup the node agent itself stopped waiting for (its context ended): no failure, so // the next start backs the module up again rather than holding it for the operator. @@ -150,6 +161,9 @@ type Env struct { // client to fill again what it refused (issue 924, the second meeting: ace, g14 and shanks on 2026-10-29 at 00:09 // UTC refused power's three new drop-ins from an engine start on, as no module settled after it). Nil tells nothing. RunsChanged func(why string) + // Now is the clock the agent compares a pull's start with the time it heard a store's holder settled by; nil is + // time.Now. + Now func() time.Time // Backoff is how long to wait before the try-th retry (1, 2, …) of an apply that failed transiently. Nil retries a // transient failure only on RetryTransient or a change of either key. Backoff func(try int) time.Duration @@ -232,11 +246,17 @@ type Agent struct { failedAgain chan string // tries counts the transient failures in a row of each module; again carries a retry whose wait ran out, and // transient asks for every transiently failed module at once. stopped is closed when Run returns. - tries map[string]int - gen uint64 - again chan againAsk - transient chan struct{} - stopped chan struct{} + tries map[string]int + // pulls is, for each module whose last apply failed on a pull from a store, what it waits on (ENG-8); settledAt and + // settledSeq are, by the module holding a store, when its settled was last heard since this agent started and how many + // have been. Under mu. + pulls map[string]*pullWait + settledAt map[string]time.Time + settledSeq map[string]int + gen uint64 + again chan againAsk + transient chan struct{} + stopped chan struct{} } type againAsk struct { @@ -276,6 +296,7 @@ func New(env Env) *Agent { } return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, applied: map[string]compNow{}, had: map[string]compNow{}, keptData: map[string]bool{}, attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}, notComposed: map[string]notComposedMark{}, + pulls: map[string]*pullWait{}, settledAt: map[string]time.Time{}, settledSeq: map[string]int{}, retry: make(chan []string, 8), reconcile: make(chan []string, 64), failedAgain: make(chan string, 8), tries: map[string]int{}, again: make(chan againAsk, 64), transient: make(chan struct{}, 1), stopped: make(chan struct{})} } @@ -1510,6 +1531,9 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseApplying)}) res := a.env.Apply(ctx, comp.C, others) a.mu.Lock() + if !res.Failed || res.Transient { + delete(a.pulls, module) + } what := fmt.Sprintf("%s at %s", module, fp) if module == NodeModule { what = "the node's own composition" @@ -1544,11 +1568,20 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo delete(a.tries, module) if res.Failed { a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev} + detail := res.Detail + again := a.notePullFailed(module, res.Pulls) + if words := a.pullWords(module); words != "" { + detail += words + } a.mu.Unlock() a.env.Say(fmt.Sprintf("applying %s failed: %s", what, res.Detail)) - a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed), Detail: res.Detail}) + a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed), Detail: detail}) + if again != "" { + a.askPullAgain(module, again) + } return } + delete(a.pulls, module) delete(a.attempt, module) delete(a.notComposed, module) // the apply says what it runs now a.running[module] = delivery.Have{FP: fp, Composed: dig, Phase: delivery.PhaseRunning} @@ -1584,6 +1617,7 @@ func (a *Agent) remove(ctx context.Context, module string, w wantNow) { return } delete(a.attempt, module) + delete(a.pulls, module) delete(a.had, module) a.running[module] = delivery.Have{Phase: delivery.PhaseRemoved} a.mu.Unlock() @@ -1794,3 +1828,107 @@ func (a *Agent) readKeptData(ctx context.Context, modules ...string) { } } } + +// pullWait is what a module whose apply failed on a pull from a store waits on: the store's holder settling (ENG-8, +// design 51 *distribution's first delivery*). +type pullWait struct { + PullFailure + // retriedFor is the count of the holder's settled the last retry answered: a settled answers once. + retriedFor int + // last is what triggered the last retry, in words, and lastAt when; the have and the journal say it. + last string + lastAt time.Time +} + +func (a *Agent) now() time.Time { + if a.env.Now != nil { + return a.env.Now() + } + return time.Now() +} + +// notePullFailed records the pull a failed apply met from a store (the first image with a holder), keeping the settled it was +// last tried again for. It answers, when the holder's settled was heard after the pull started and not yet answered, the +// words of the retry to make at once (the race just before `settled`: ENG-11), else "". a.mu is held. +func (a *Agent) notePullFailed(module string, pulls []PullFailure) string { + var found *PullFailure + for i := range pulls { + if pulls[i].Holder != "" { + found = &pulls[i] + break + } + } + if found == nil { + delete(a.pulls, module) + return "" + } + pw := &pullWait{PullFailure: *found} + if before := a.pulls[module]; before != nil { + pw.retriedFor, pw.last, pw.lastAt = before.retriedFor, before.last, before.lastAt + } + a.pulls[module] = pw + holder := found.Holder + if seq := a.settledSeq[holder]; seq > pw.retriedFor && a.settledAt[holder].After(found.Started) { + return a.markPullRetry(pw, seq, fmt.Sprintf("%s settled at %s, after the pull of %s began at %s", holder, + a.settledAt[holder].Format("15:04:05"), found.Image, found.Started.Format("15:04:05"))) + } + return "" +} + +// markPullRetry notes that pw is tried again for the holder's settled number seq, and answers the words that say so. a.mu +// is held. +func (a *Agent) markPullRetry(pw *pullWait, seq int, why string) string { + pw.retriedFor, pw.last, pw.lastAt = seq, why, a.now() + return why +} + +// pullWords is what a failed have says of the wait: whose settling it is tried again on and what triggered the last retry. +// "" for a failure that waits on no holder. a.mu is held. +func (a *Agent) pullWords(module string) string { + pw := a.pulls[module] + if pw == nil { + return "" + } + words := fmt.Sprintf("; it is tried again once %s has settled", pw.Holder) + if pw.last != "" { + words += "; last tried again at " + pw.lastAt.Format("15:04:05") + ": " + pw.last + } + return words +} + +// askPullAgain asks the apply of a module again, because why, as a person's verb compose does (RetryFailed): said in the +// journal, never blocking. +func (a *Agent) askPullAgain(module, why string) { + select { + case a.failedAgain <- module: + a.env.Say(fmt.Sprintf("%s: its image could not be pulled from the store; applied again because %s", module, why)) + default: + a.env.Say(fmt.Sprintf("%s: its image could not be pulled from the store, and %s, but the engine has too many retries "+ + "asked; ask it with compose", module, why)) + } +} + +// Settled is the status view's word that a module has settled on some node (ENG-11): the engine's hearing tells it for every +// module it hears. When the module holds a store a failed apply of this node waits on, that apply is tried again once for +// this settled; a settled of any other module starts nothing. A pull that began after this settled, or has failed before it +// and is answered, is not tried again for it (notePullFailed). +func (a *Agent) Settled(holder string) { + a.mu.Lock() + a.settledSeq[holder]++ + a.settledAt[holder] = a.now() + seq := a.settledSeq[holder] + var asks []string + var modules []string + for module, pw := range a.pulls { + att, failed := a.attempt[module] + if pw.Holder != holder || !failed || att.Held || att.Transient || pw.retriedFor >= seq { + continue + } + modules = append(modules, module) + asks = append(asks, a.markPullRetry(pw, seq, fmt.Sprintf("%s settled", holder))) + } + a.mu.Unlock() + for i, module := range modules { + a.askPullAgain(module, asks[i]) + } +} diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index e2fed75..4592d48 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -8,6 +8,7 @@ import ( "slices" "strings" "sync" + "sync/atomic" "testing" "time" @@ -115,10 +116,14 @@ type bed struct { // transient marks failWith as the bus not answering; failTimes > 0 fails only the first that many applies. transient bool failTimes int - backup func(module string) Backup - kept map[string][]json.RawMessage - others [][]delivery.Composed - said []string + // pullsFn, when set, is the images a failed apply could not pull (called as the apply fails); during runs at the start of + // each apply, before it answers. + pullsFn func() []PullFailure + during func() + backup func(module string) Backup + kept map[string][]json.RawMessage + others [][]delivery.Composed + said []string // settled is each module Settled was told, and settledRan whether what runs held it then. settled []string settledRan []bool @@ -130,6 +135,9 @@ func newBed(t *testing.T) *bed { b := &bed{t: t, wants: newFeed(), comps: newFeed(), haves: &haves{}} b.env = Env{Node: node, Wants: b.wants, Composed: b.comps, Haves: b.haves, Apply: func(_ context.Context, c delivery.Composed, others []delivery.Composed) Result { + if b.during != nil { + b.during() + } b.mu.Lock() defer b.mu.Unlock() b.others = append(b.others, others) @@ -139,7 +147,11 @@ func newBed(t *testing.T) *bed { } b.applied = append(b.applied, name) if b.failWith != "" && (b.failTimes == 0 || len(b.applied) <= b.failTimes) { - return Result{Failed: true, Transient: b.transient, Detail: b.failWith} + res := Result{Failed: true, Transient: b.transient, Detail: b.failWith} + if b.pullsFn != nil { + res.Pulls = b.pullsFn() + } + return res } return Result{} }, @@ -461,6 +473,106 @@ func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) { t.Errorf("a module running its want was applied again (%d applies)", n) } }) + // design 51 (ENG-8, ENG-11): a pull from a store that failed is tried again when the store's holder settles, once per + // settled, never on a clock; the have says whom it waits on and what triggered the last retry. + t.Run("a pull from a store waits for its holder to settle", func(t *testing.T) { + b := newBed(t) + var clock atomic.Int64 + now := func() time.Time { return time.Unix(1_000_000, clock.Load()) } + tick := func() time.Time { clock.Add(int64(time.Second)); return now() } + b.env.Now = now + b.env.Backoff = func(int) time.Duration { return time.Millisecond } + b.failWith = "pulling image reg/x@sha256:1: manifest unknown" + b.pullsFn = func() []PullFailure { + return []PullFailure{{Image: "reg/x@sha256:1", Seat: "mesh-artifact-store", Holder: "distribution", Started: tick()}, + {Image: "public/y@sha256:2"}} + } + stop := b.run() + defer stop() + fp := fpOf("a") + ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1")) + wk, wraw := wantFor(t, "web", fp, dig, "") + b.comps.put(ck, craw) + b.wants.put(wk, wraw) + eventually(t, "not applied", func() bool { return len(b.appliedList()) == 1 }) + eventually(t, "the have does not say whom it waits on", func() bool { + h := b.haves.last("web") + return h.Phase == string(delivery.PhaseFailed) && strings.Contains(h.Detail, "tried again once distribution has settled") + }) + time.Sleep(100 * time.Millisecond) + if n := len(b.appliedList()); n != 1 { + t.Fatalf("a failed pull was tried again by the clock (%d applies)", n) + } + // A settled of another module starts nothing. + b.ag.Settled("gitea") + time.Sleep(50 * time.Millisecond) + if n := len(b.appliedList()); n != 1 { + t.Fatalf("a settled of another module applied the module (%d applies)", n) + } + // The holder's settled: tried again once, whose pull fails again after it (started later): failed again, no more. + tick() + b.ag.Settled("distribution") + eventually(t, "not tried again on the holder's settled", func() bool { return len(b.appliedList()) == 2 }) + eventually(t, "the have does not say what triggered the retry", func() bool { + h := b.haves.last("web") + return h.Phase == string(delivery.PhaseFailed) && strings.Contains(h.Detail, "last tried again at") && strings.Contains(h.Detail, "distribution settled") + }) + time.Sleep(100 * time.Millisecond) + if n := len(b.appliedList()); n != 2 { + t.Fatalf("a second failure after the same settled started a further apply (%d applies)", n) + } + // The next settled answers once more, and the pull works: the module runs, and waits on nothing. + b.mu.Lock() + b.failWith = "" + b.mu.Unlock() + tick() + b.ag.Settled("distribution") + eventually(t, "not tried again on the next settled", func() bool { return len(b.appliedList()) == 3 }) + eventually(t, "not running", func() bool { return b.haves.last("web").Phase == string(delivery.PhaseRunning) }) + b.ag.Settled("distribution") + time.Sleep(50 * time.Millisecond) + if n := len(b.appliedList()); n != 3 { + t.Fatalf("a running module was applied again by a settled (%d applies)", n) + } + }) + // The race just before `settled`: the holder's settled is heard while the pull is in flight, and the pull fails after it; + // applied again at once, once, and a settled heard before the pull began applies nothing at once. + t.Run("a settled heard while the pull ran is not missed", func(t *testing.T) { + b := newBed(t) + var clock atomic.Int64 + now := func() time.Time { return time.Unix(1_000_000, clock.Load()) } + tick := func() time.Time { clock.Add(int64(time.Second)); return now() } + b.env.Now = now + b.failWith = "pulling image reg/x@sha256:1: connection refused" + began := tick() + first := true + b.during = func() { + if first { + first = false + tick() + b.ag.Settled("distribution") // the holder settles while this apply's pull is in flight + } + } + b.pullsFn = func() []PullFailure { + started := began + if len(b.applied) > 1 { + started = tick() // the retry's pull begins after the settled + } + return []PullFailure{{Image: "reg/x@sha256:1", Seat: "mesh-artifact-store", Holder: "distribution", Started: started}} + } + stop := b.run() + defer stop() + fp := fpOf("a") + ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1")) + wk, wraw := wantFor(t, "web", fp, dig, "") + b.comps.put(ck, craw) + b.wants.put(wk, wraw) + eventually(t, "the settled heard during the pull was missed", func() bool { return len(b.appliedList()) == 2 }) + time.Sleep(100 * time.Millisecond) + if n := len(b.appliedList()); n != 2 { + t.Fatalf("the race retry failing again started a further apply (%d applies)", n) + } + }) // The bus did not answer for a moment (mesh-vault on the bus node, 2026-10-25 23:04:24): tried again with no new // revision of either key, by the backoff or by RetryTransient once the engine knows the bus is back; the have says // retrying meanwhile, never failed. diff --git a/internal/apply/apply.go b/internal/apply/apply.go index d97ff1d..260e58a 100644 --- a/internal/apply/apply.go +++ b/internal/apply/apply.go @@ -3481,16 +3481,17 @@ func ensureImage(ctx context.Context, cri, image string, run Runner) error { // already hold — built here, or handed over. There is no registry that answers for it, so // pulling would fail somewhere that names a network problem instead of a missing image. if strings.HasPrefix(image, "sha256:") { - return &PullError{Image: image, Err: fmt.Errorf( + return &PullError{Image: image, Started: time.Now(), Err: fmt.Errorf( "image %s is not on this machine, and an image named by its own digest cannot be "+ "fetched: nothing serves it. Build it here, or load it, before applying this", image)} } + started := time.Now() if _, err := run(ctx, cri, "pull", image); err != nil { - return &PullError{Image: image, Err: fmt.Errorf("pulling image %s: %w", image, err)} + return &PullError{Image: image, Started: started, Err: fmt.Errorf("pulling image %s: %w", image, err)} } if _, err := run(ctx, cri, "image", "inspect", image); err != nil { - return &PullError{Image: image, Err: fmt.Errorf("image %s is not present after pulling it: %w", image, err)} + return &PullError{Image: image, Started: time.Now(), Err: fmt.Errorf("image %s is not present after pulling it: %w", image, err)} } return nil } @@ -4405,7 +4406,33 @@ func pullFirst(ctx context.Context, ordered []declaration.Resource, d *declarati // pull from another failed apply (ENG-8). type PullError struct { Image string - Err error + // Started is when the pull began, which the node agent compares with when a store's holder settled (ENG-11). + Started time.Time + Err error +} + +// PullErrors are the pull errors of every failure an apply reports, the first among them and the rest in Failures, in the +// order they happened: what the node agent tells a failed pull from another failed apply by (ENG-8). Nil for an error +// that is no apply's, or whose failures met none. +func PullErrors(err error) []*PullError { + var out []*PullError + var e *Error + if errors.As(err, &e) { + for _, f := range e.Failures { + var p *PullError + if errors.As(f.Err, &p) { + out = append(out, p) + } + } + if len(out) > 0 { + return out + } + } + var p *PullError + if errors.As(err, &p) { + out = append(out, p) + } + return out } func (e *PullError) Error() string { return e.Err.Error() }