Merge pull request 'The node-engine fetches every artifact before it replaces anything, and a failed apply no longer claims the old composition runs (issue 1966, recovery)' (#215) from fix/1966-fetch-before-replace into main

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]

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) {

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

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)
+	// What the apply fetched before anything was replaced (fetchFirst), else fetched now.
-	if err != nil {
+	body, err := held, error(nil)
-		return out, err
+	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[:])

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
+}

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)
+	// What the apply fetched before anything was replaced (fetchFirst), else fetched now.
-	if err != nil {
+	body, err := held, error(nil)
-		return out, err
+	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[:])

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) {