diff --git a/internal/agent/agent.go b/internal/agent/agent.go index d1e8bb5..d92bed2 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -1372,7 +1372,9 @@ func (a *Agent) settleOne(ctx context.Context, module string) { } a.mu.Lock() r, runs := a.running[module] - runsWant := runs && r.Fp == w.Want.Fp && r.Composed == w.Want.Composed && r.Phase != nodeengineclient.PhaseRemoved + // A module whose last apply failed runs no composition whole, whatever it ran before (ranFailed). + runsWant := runs && r.Fp == w.Want.Fp && r.Composed == w.Want.Composed && r.Phase != nodeengineclient.PhaseRemoved && + r.Phase != nodeengineclient.PhaseFailed att, tried := a.attempt[module] same := tried && att.FP == w.Want.Fp && att.Composed == w.Want.Composed if same && att.Held { @@ -1598,6 +1600,9 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo if !res.Failed || res.Transient { delete(a.pulls, module) } + if res.Failed { + a.ranFailed(module) + } what := fmt.Sprintf("%s at %s", module, fp) if module == NodeModule { what = "the node's own composition" @@ -1662,6 +1667,16 @@ 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(nodeengineclient.PhaseRunning), Detail: res.Detail}) } +// ranFailed marks what runs of a module as failed once an apply of it failed (novox/hq issue 1966, 2026-11-08): a failed +// apply may have replaced some of its resources (distribution's store container) before it stopped, so what ran is no +// longer whole, and a want of it again (a put back) is applied rather than taken as running. Called with a.mu held. +func (a *Agent) ranFailed(module string) { + if r, ok := a.running[module]; ok && r.Phase != nodeengineclient.PhaseRemoved { + r.Phase = nodeengineclient.PhaseFailed + a.running[module] = r + } +} + func (a *Agent) remove(ctx context.Context, module string, w wantNow) { a.mu.Lock() att := a.attempt[module] diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 0b3738d..2b778d6 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -487,6 +487,45 @@ func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) { t.Errorf("a module running its want was applied again (%d applies)", n) } }) + // Issue 1966 (2026-11-08 23:47 UTC): a failed apply of distribution's new build had replaced its store container, and + // the put back to the build that ran before applied nothing, judged as running. A module whose apply failed runs no + // composition whole: the want of what ran before is applied again. + t.Run("1966: after a failed apply, the composition that ran is applied again when wanted", func(t *testing.T) { + b := newBed(t) + stop := b.run() + defer stop() + fpA, fpB := fpOf("a"), fpOf("b") + ckA, crawA, digA := composition(t, "web", fpA, false, file("web", "x", "1")) + ckB, crawB, digB := composition(t, "web", fpB, false, file("web", "x", "2")) + b.comps.put(ckA, crawA) + b.comps.put(ckB, crawB) + wk, wrawA := wantFor(t, "web", fpA, digA, "") + _, wrawB := wantFor(t, "web", fpB, digB, "") + b.wants.put(wk, wrawA) + b.idle() + if n := len(b.appliedList()); n != 1 || b.haves.last("web").Phase != string(nodeengineclient.PhaseRunning) { + t.Fatalf("the first build was not applied and running (%d applies, have %+v)", n, b.haves.last("web")) + } + b.mu.Lock() + b.failWith = "distribution.program: the store does not answer" + b.mu.Unlock() + b.wants.put(wk, wrawB) + b.idle() + if n := len(b.appliedList()); n != 2 || b.haves.last("web").Phase != string(nodeengineclient.PhaseFailed) { + t.Fatalf("the second build's apply did not fail (%d applies, have %+v)", n, b.haves.last("web")) + } + b.mu.Lock() + b.failWith = "" + b.mu.Unlock() + b.wants.put(wk, wrawA) // the put back + b.idle() + if n := len(b.appliedList()); n != 3 { + t.Fatalf("the put back to the build that ran applied nothing (%d applies): a failed apply left it said as running", n) + } + if h := b.haves.last("web"); h.FP != fpA || h.Composed != digA || h.Phase != string(nodeengineclient.PhaseRunning) { + t.Errorf("after the put back the have is %+v, want the first build running", h) + } + }) // 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) { diff --git a/internal/apply/apply.go b/internal/apply/apply.go index 260e58a..a6c00f9 100644 --- a/internal/apply/apply.go +++ b/internal/apply/apply.go @@ -708,10 +708,24 @@ func applyScoped( ordered = append(afterWhatTheyRead(guard), afterWhatTheyRead(rest)...) guardFirst = len(guard) } + // Modules whose own step did not complete. What follows *within such a module* is not attempted; + // the rest of the machine is (novox/hq ADR 0136). + gated := map[string]bool{} + // Why each gated module was gated, said on every resource of it that was skipped. + gatedWhy := map[string]string{} orphansRemoved := false removeOrphans := func() error { orphansRemoved = true for _, orphan := range orphans { + // **Nothing of a module whose image or artifact could not be had is removed either** (novox/hq issue 1966): + // a removal is as much a change of what it runs as a replacement, and the module keeps running what it ran. + if m := orphanModule(orphan); m != "" && gated[m] { + report.Outcomes = append(report.Outcomes, Outcome{ + ID: orphan.ID, Type: orphan.Type, Target: orphan.Target, Action: "kept", Detail: gatedWhy[m], + }) + log(fmt.Sprintf(" kept %s (%s): %s", orphan.ID, orphan.Target, gatedWhy[m])) + continue + } if err := removeOrphan(orphan); err != nil { return err } @@ -797,11 +811,6 @@ func applyScoped( var failures []*Error // Which resources applied without error in this run, for the move-out pass after the loop (issue 1024). appliedOK := map[string]bool{} - // Modules whose own step did not complete. What follows *within such a module* is not attempted; - // the rest of the machine is (novox/hq ADR 0136). - gated := map[string]bool{} - // Why each gated module was gated, said on every resource of it that was skipped. - gatedWhy := map[string]string{} gatedModules := func() []string { out := make([]string, 0, len(gated)) for m := range gated { @@ -946,6 +955,11 @@ func applyScoped( // follows in the module is skipped as it is for any step that did not complete. in.pulled = map[string]bool{} pullFirst(ctx, ordered, d, in.pulled, run, &failures, gated, gatedWhy, log) + // And every archive and process, fetched before anything is replaced (issue 1966). Its failures are counted with the + // others once the resources are through, as an identity's are: here, before the guard, they would read as the guard + // not being up, and nothing of the machine would be applied. + var fetchFailures []*Error + fetchFirst(ctx, ordered, d, &in, &fetchFailures, gated, gatedWhy, log) for i, resource := range ordered { if !orphansRemoved && i == guardFirst { // **Only a guard that is up may let the filter go.** Removing the derived filter's @@ -1304,6 +1318,7 @@ func applyScoped( } failures = append(failures, identityFailures...) + failures = append(failures, fetchFailures...) // A directory or archive that says move-out keeps only what it knows (issue 1024, novox/hq ADR 0422). movedOut, leftIn, moveFailures := moveOutPass(d, known, appliedOK, log) @@ -1579,9 +1594,17 @@ func applyOne(ctx context.Context, sys system.System, r declaration.Resource, ru case *declaration.User: return applyUser(ctx, sys, res, run, previous, in.groups, in.handed) case *declaration.Archive: - return applyArchive(ctx, res, previous, in.away, journalDir(in.windows)) + held, err := in.artifact(res.Identity()) + if err != nil { + return begin(r), err + } + return applyArchive(ctx, res, previous, held, in.away, journalDir(in.windows)) case *declaration.Process: - return applyProcess(ctx, res, run, changed, previous, in.away) + held, err := in.artifact(res.Identity()) + if err != nil { + return begin(r), err + } + return applyProcess(ctx, res, run, changed, previous, held, in.away) case *declaration.Action: return applyAction(ctx, res, run) case *declaration.Network: @@ -2769,6 +2792,9 @@ type inputs struct { // pulled is every image this apply found on the machine, or pulled, before any container was replaced (pullFirst): // applyContainer does not ask for it again. pulled map[string]bool + // fetched is every archive's and process's bytes this apply fetched before anything was replaced (fetchFirst), each + // taken by its applier once (artifact). + fetched map[string]fetchedArtifact } // fileDigest is what a file the container reads holds, by digest. @@ -4354,6 +4380,14 @@ func ownerText(info os.FileInfo) string { return fmt.Sprintf("%d:%d", uid, gid) } +// orphanModule is the module a record going belongs to, by its id, else the module whose want it was applied for. +func orphanModule(a store.Applied) string { + if m, ok := declaration.ModuleGuess(a.ID); ok { + return m + } + return a.Module +} + // pullFirst finds on the machine, or pulls, the image of each container the declaration runs before any is replaced // (ENG-7) for a module of two containers or more. A container that is run once is left to its marker, and a runtime that // does not answer to applyContainer, which says so. The first image that cannot be had fails its container with the image named and gates its module, so none of diff --git a/internal/apply/archive.go b/internal/apply/archive.go index aeb8cd5..bb6c343 100644 --- a/internal/apply/archive.go +++ b/internal/apply/archive.go @@ -35,7 +35,7 @@ import ( // enough for a desktop theme and small enough to notice. const maxArchive = 512 << 20 -func applyArchive(ctx context.Context, r *declaration.Archive, previous store.Applied, away *storeAway, journals string) (Outcome, error) { +func applyArchive(ctx context.Context, r *declaration.Archive, previous store.Applied, held []byte, away *storeAway, journals string) (Outcome, error) { out := begin(r) out.Action = "unchanged" recorded := previous.Wrote != "" @@ -44,9 +44,12 @@ func applyArchive(ctx context.Context, r *declaration.Archive, previous store.Ap return out, err } - body, err := fetchMinding(ctx, r.Source, away) - if err != nil { - return out, err + // What the apply fetched before anything was replaced (fetchFirst), else fetched now. + body, err := held, error(nil) + if body == nil { + if body, err = fetchMinding(ctx, r.Source, away); err != nil { + return out, err + } } sum := sha256.Sum256(body) got := "sha256:" + hex.EncodeToString(sum[:]) diff --git a/internal/apply/fetch_first.go b/internal/apply/fetch_first.go new file mode 100644 index 0000000..ad91ec0 --- /dev/null +++ b/internal/apply/fetch_first.go @@ -0,0 +1,102 @@ +package apply + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + + "git.novox.be/novox/mesh-sdk/go/declaration" +) + +// **Every artifact an apply fetches is in hand before anything is replaced** (novox/hq issue 1966, 2026-11-08 23:25 UTC). +// +// An archive and a process are fetched from the artifact store on their own turn, and a container is replaced on its +// own: so a module whose container serves the store (distribution) had that container replaced, cut off from where it +// was reached, and then failed to fetch its program from the store it had just cut off, left half applied with the +// store unreachable from every other machine. pullFirst does this for a module's images (ENG-7); fetchFirst does it for +// its archives and processes, before any of the apply's resources is applied or any orphan removed. +// +// A fetch that fails (the store does not answer for the apply's patience, or what arrived is not what was declared) +// gates the resource's module, as a pull that fails does: nothing of the module is replaced or removed, the failure names +// the artifact, and the rest of the machine is applied. A resource of no module keeps its failure for its own turn. + +// fetchedArtifact is one archive's or process's bytes, fetched and checked before the apply began, or why it was not. +type fetchedArtifact struct { + body []byte + err error +} + +// artifactOf is the archive's or process's source and digest, and whether the resource is one the apply fetches. +func artifactOf(r declaration.Resource) (source, digest string, ok bool) { + switch res := r.(type) { + case *declaration.Archive: + return res.Source, res.Digest, res.Source != "" + case *declaration.Process: + return res.Source, res.Digest, res.Source != "" + } + return "", "", false +} + +// fetchFirst fetches each archive and process the declaration applies, checks each against its declared digest, and +// keeps the bytes for its turn (in.fetched). It is called before the loop that applies, after pullFirst. +func fetchFirst(ctx context.Context, ordered []declaration.Resource, d *declaration.Declaration, in *inputs, + failures *[]*Error, gated map[string]bool, gatedWhy map[string]string, log func(string)) { + in.fetched = map[string]fetchedArtifact{} + bySource := map[string]fetchedArtifact{} + for _, resource := range ordered { + source, digest, isArtifact := artifactOf(resource) + if !isArtifact { + continue + } + module, ours := d.ModuleOfID(resource.Identity()) + if ours && gated[module] { + continue + } + // On an adopted node, what a module not yet taken declares may be held as found and never fetched (hold.go): its + // own turn fetches it, as before. + if d.Adoption != nil { + if _, untaken := d.Adoption.UntakenModuleOf(resource.Identity()); untaken || runsIn(resource) != "" { + continue + } + } + got, seen := bySource[source] + if !seen { + got.body, got.err = fetchMinding(ctx, source, in.away) + bySource[source] = got + } + if got.err == nil { + sum := sha256.Sum256(got.body) + if have := "sha256:" + hex.EncodeToString(sum[:]); have != digest { + got = fetchedArtifact{err: fmt.Errorf("%s was declared as %s and what arrived is %s; nothing was unpacked", + source, digest, have)} + } + } + if got.err == nil { + in.fetched[resource.Identity()] = got + continue + } + if !ours { + // Nothing to gate: the resource fails on its own turn with this error. + in.fetched[resource.Identity()] = got + continue + } + err := fmt.Errorf("not replacing anything of its module, which keeps running what it ran: %w", got.err) + *failures = append(*failures, &Error{Resource: resource.Identity(), Err: err}) + log(fmt.Sprintf(" failed %s (%s): %v", resource.Identity(), resource.Target(), err)) + gated[module] = true + gatedWhy[module] = "its artifact " + source + " could not be fetched, so nothing of it was replaced " + + "(novox/hq issue 1966)" + } +} + +// artifact is the bytes fetchFirst holds for a resource, taken once so they are not kept past their turn; nil and no +// error when it holds none (an apply that did not fetch first: its applier fetches itself). +func (in inputs) artifact(id string) ([]byte, error) { + got, ok := in.fetched[id] + if !ok { + return nil, nil + } + delete(in.fetched, id) + return got.body, got.err +} diff --git a/internal/apply/process.go b/internal/apply/process.go index 121b683..5e05d99 100644 --- a/internal/apply/process.go +++ b/internal/apply/process.go @@ -40,13 +40,16 @@ var daemonRoot = "/var/lib/mesh/daemons" var unitDir = "/etc/systemd/system" func applyProcess(ctx context.Context, r *declaration.Process, run Runner, - changed map[string]bool, previous store.Applied, away *storeAway) (Outcome, error) { + changed map[string]bool, previous store.Applied, held []byte, away *storeAway) (Outcome, error) { out := begin(r) out.Action = "unchanged" - body, err := fetchMinding(ctx, r.Source, away) - if err != nil { - return out, err + // What the apply fetched before anything was replaced (fetchFirst), else fetched now. + body, err := held, error(nil) + if body == nil { + if body, err = fetchMinding(ctx, r.Source, away); err != nil { + return out, err + } } sum := sha256.Sum256(body) got := "sha256:" + hex.EncodeToString(sum[:]) diff --git a/internal/apply/replace_image_test.go b/internal/apply/replace_image_test.go index 99f6b5b..fa48c83 100644 --- a/internal/apply/replace_image_test.go +++ b/internal/apply/replace_image_test.go @@ -3,6 +3,8 @@ package apply import ( "context" "errors" + "net/http" + "net/http/httptest" "strings" "testing" @@ -104,6 +106,64 @@ func TestARunningContainerIsKeptWhenItsReplacementImageCannotBeObtained(t *testi t.Errorf("%s: the failure does not name %s: %v", c.name, second, err) } } + + // Issue 1966 (2026-11-08 23:25 UTC): a module whose container serves the artifact store had it replaced, and then + // could not fetch its program from the store it had cut off. Every archive and process is fetched before anything of + // the module is replaced or removed: with the store away, its container and its orphan stay as they were, and the + // failure names the program's source. + t.Run("an artifact that cannot be fetched replaces nothing of its module", func(t *testing.T) { + gone := httptest.NewServer(http.NotFoundHandler()) + source := gone.URL + "/program.tar.gz" + gone.Close() // the store is cut off: refused + was := storeAwayAtMost + storeAwayAtMost = 0 // no patience: the refusal fails at once, without waiting + t.Cleanup(func() { storeAwayAtMost = was }) + d := parseTrusted(t, `{"declaration":1,"resources":[ + {"id":"distribution.store","type":"container","name":"store","image":"`+pinned+`","env":{"X":"new"}}, + {"id":"distribution.program","type":"process","name":"program","source":"`+source+`", + "digest":"sha256:`+strings.Repeat("0", 64)+`","run":["program"]} + ]}`) + known := store.State{} + known.Record(store.Applied{ID: "distribution.old", Type: string(declaration.TypeContainer), Target: "old", + Module: "distribution"}) + var steps []string + removed := map[string]bool{} + run := func(ctx context.Context, name string, args ...string) (string, error) { + if name == "systemctl" { + steps = append(steps, "systemctl "+strings.Join(args, " ")) + return "", nil + } + args = oldShape(name, args) + switch args[0] { + case "info": + return "27.0\n", nil + case "container": + if removed[args[len(args)-1]] { + return "", errors.New("no such container") + } + return "true\tsome-older-spec", nil + case "rm", "rename", "stop", "run": + steps = append(steps, args[0]+" "+strings.Join(args[1:], " ")) + if args[0] == "rm" { + removed[args[len(args)-1]] = true + } + } + return "", nil + } + report, after, err := Apply(context.Background(), archHost(t), d, known, store.OriginCarried, run, nil, nil) + if len(steps) > 0 { + t.Errorf("the module was touched though its program could not be fetched: %q", steps) + } + if err == nil || !strings.Contains(err.Error(), source) { + t.Errorf("the failure does not name the program's source %s: %v", source, err) + } + if _, still := after.Find("distribution.old"); !still { + t.Error("the module's orphan was removed though its program could not be fetched") + } + if o := outcomeOf(report, "distribution.store"); o.Action != "skipped" { + t.Errorf("the container's outcome is %+v, want skipped", o) + } + }) } func TestAReplacementImageIsPresentBeforeTheRunningContainerIsRemoved(t *testing.T) {