Fetch every archive and process of an apply before anything is replaced (issue 1966, recovery)
On 2026-11-08 23:25 UTC the node-engine replaced distribution's store container and only then fetched distribution's program from the store it had just cut off: the module was left half applied and the artifact store unreachable from every other machine. An apply now fetches and checks every archive and process first, as it already pulls a module's images first; one that cannot be had gates its module, so none of its resources is replaced and none of its orphans removed. Claude-Session: https://claude.ai/code/session_01NcbgfFfQ4Z9UE5jK32PmE5
This commit is contained in:
+41
-7
@@ -708,10 +708,24 @@ func applyScoped(
|
|||||||
ordered = append(afterWhatTheyRead(guard), afterWhatTheyRead(rest)...)
|
ordered = append(afterWhatTheyRead(guard), afterWhatTheyRead(rest)...)
|
||||||
guardFirst = len(guard)
|
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
|
orphansRemoved := false
|
||||||
removeOrphans := func() error {
|
removeOrphans := func() error {
|
||||||
orphansRemoved = true
|
orphansRemoved = true
|
||||||
for _, orphan := range orphans {
|
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 {
|
if err := removeOrphan(orphan); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -797,11 +811,6 @@ func applyScoped(
|
|||||||
var failures []*Error
|
var failures []*Error
|
||||||
// Which resources applied without error in this run, for the move-out pass after the loop (issue 1024).
|
// Which resources applied without error in this run, for the move-out pass after the loop (issue 1024).
|
||||||
appliedOK := map[string]bool{}
|
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 {
|
gatedModules := func() []string {
|
||||||
out := make([]string, 0, len(gated))
|
out := make([]string, 0, len(gated))
|
||||||
for m := range 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.
|
// follows in the module is skipped as it is for any step that did not complete.
|
||||||
in.pulled = map[string]bool{}
|
in.pulled = map[string]bool{}
|
||||||
pullFirst(ctx, ordered, d, in.pulled, run, &failures, gated, gatedWhy, log)
|
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 {
|
for i, resource := range ordered {
|
||||||
if !orphansRemoved && i == guardFirst {
|
if !orphansRemoved && i == guardFirst {
|
||||||
// **Only a guard that is up may let the filter go.** Removing the derived filter's
|
// **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, identityFailures...)
|
||||||
|
failures = append(failures, fetchFailures...)
|
||||||
|
|
||||||
// A directory or archive that says move-out keeps only what it knows (issue 1024, novox/hq ADR 0422).
|
// 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)
|
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:
|
case *declaration.User:
|
||||||
return applyUser(ctx, sys, res, run, previous, in.groups, in.handed)
|
return applyUser(ctx, sys, res, run, previous, in.groups, in.handed)
|
||||||
case *declaration.Archive:
|
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:
|
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:
|
case *declaration.Action:
|
||||||
return applyAction(ctx, res, run)
|
return applyAction(ctx, res, run)
|
||||||
case *declaration.Network:
|
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):
|
// 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.
|
// applyContainer does not ask for it again.
|
||||||
pulled map[string]bool
|
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.
|
// 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)
|
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
|
// 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
|
// (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
|
// 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
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ import (
|
|||||||
// enough for a desktop theme and small enough to notice.
|
// enough for a desktop theme and small enough to notice.
|
||||||
const maxArchive = 512 << 20
|
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 := begin(r)
|
||||||
out.Action = "unchanged"
|
out.Action = "unchanged"
|
||||||
recorded := previous.Wrote != ""
|
recorded := previous.Wrote != ""
|
||||||
@@ -44,9 +44,12 @@ func applyArchive(ctx context.Context, r *declaration.Archive, previous store.Ap
|
|||||||
return out, err
|
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)
|
sum := sha256.Sum256(body)
|
||||||
got := "sha256:" + hex.EncodeToString(sum[:])
|
got := "sha256:" + hex.EncodeToString(sum[:])
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -40,13 +40,16 @@ var daemonRoot = "/var/lib/mesh/daemons"
|
|||||||
var unitDir = "/etc/systemd/system"
|
var unitDir = "/etc/systemd/system"
|
||||||
|
|
||||||
func applyProcess(ctx context.Context, r *declaration.Process, run Runner,
|
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 := begin(r)
|
||||||
out.Action = "unchanged"
|
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)
|
sum := sha256.Sum256(body)
|
||||||
got := "sha256:" + hex.EncodeToString(sum[:])
|
got := "sha256:" + hex.EncodeToString(sum[:])
|
||||||
|
|||||||
@@ -3,6 +3,8 @@ package apply
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
@@ -104,6 +106,64 @@ func TestARunningContainerIsKeptWhenItsReplacementImageCannotBeObtained(t *testi
|
|||||||
t.Errorf("%s: the failure does not name %s: %v", c.name, second, err)
|
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) {
|
func TestAReplacementImageIsPresentBeforeTheRunningContainerIsRemoved(t *testing.T) {
|
||||||
|
|||||||
Reference in New Issue
Block a user