Reconcile a module from the machine's facts at engine start and at each edge of a declared-running resource going down (issue 1957, ADR 0568)

A reboot left the have running, so a unit that failed to start at boot was never started again and nothing said so. Baseline now asks the machine; the judge's look calls Agent.Reconcile once per down edge, leaving held and unwanted modules alone.

cmd/mesh-host/healthowed_test.go

@@ -69,7 +69,7 @@ func TestAModuleWithNothingLongRunningIsOkWithinALookOfItsApply(t *testing.T) {
 	applying := fact
 	applying.Phase = string(delivery.PhaseApplying)
 	queue.NoteModule(ctx, applying)
-	go judgeWhatRuns(ctx, j, queue, func(string) {}, nil)
+	go judgeWhatRuns(ctx, j, queue, func(string) {}, nil, nil)
 	// The facts look inside the apply, while the module is still applying.
 	queue.SayFacts(ctx, link.Report{Health: &link.Health{Contract: link.LivenessContract, At: time.Now().UTC()}})
 	if h, _ := bus.have("n1.xorg"); h.Health != delivery.HealthUnknown {

cmd/mesh-host/inactive.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"context"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/novox/mesh-host/internal/apply"
+	"github.com/novox/mesh-host/internal/liveness"
+)
+
+// heldBy is the containers an open maintenance window holds still, by runtime name (ADR 0189), at now.
+func heldBy(windows *apply.Windows) func(time.Time) map[string]bool {
+	return func(now time.Time) map[string]bool {
+		held := map[string]bool{}
+		for _, w := range windows.Open(now) {
+			for _, name := range w.Holds {
+				held[name] = true
+			}
+		}
+		return held
+	}
+}
+
+// inactiveIn is what the agent asks at its start of a module it believes runs (ADR 0568 §1): whether the machine shows a
+// long-running resource the composition declares running not active, read from the service manager and the container
+// runtime and from no record. A resource held by an open maintenance window, or waiting for a library, is not counted;
+// one that cannot be read is not counted either (unknown is not inactive), and neither is one the manager is restarting.
+func inactiveIn(rt liveness.Runtime, held func(time.Time) map[string]bool) func(context.Context, delivery.Composed) bool {
+	return func(ctx context.Context, c delivery.Composed) bool {
+		d, err := declarationOf(c)
+		if err != nil {
+			return false
+		}
+		var holding map[string]bool
+		if held != nil {
+			holding = held(time.Now())
+		}
+		return anyInactive(ctx, rt, liveness.LongRunning(d, withWaiting(nil, apply.WaitingOnLibrary(d))), holding)
+	}
+}
+
+// anyInactive reads each group of resources once (the containers, each service manager's units) and says whether any is
+// not active.
+func anyInactive(ctx context.Context, rt liveness.Runtime, resources []liveness.Resource, held map[string]bool) bool {
+	type group struct {
+		kind, scope, user string
+		targets           []string
+	}
+	var groups []*group
+	at := map[string]*group{}
+	for _, r := range resources {
+		if r.Kind == liveness.KindContainer && held[r.Target] {
+			continue
+		}
+		g := &group{kind: r.Kind, scope: r.Scope, user: r.User}
+		if r.Kind == liveness.KindContainer {
+			g.scope, g.user = "", ""
+		}
+		key := g.kind + "/" + g.scope + "/" + g.user
+		if r.Kind != liveness.KindContainer {
+			key = "units/" + g.scope + "/" + g.user
+		}
+		if at[key] == nil {
+			at[key] = g
+			groups = append(groups, g)
+		}
+		at[key].targets = append(at[key].targets, r.Target)
+	}
+	for _, g := range groups {
+		var got map[string]liveness.Observed
+		var err error
+		if g.kind == liveness.KindContainer {
+			got, err = rt.Containers(ctx, g.targets)
+		} else {
+			got, err = rt.Units(ctx, g.scope, g.user, g.targets)
+		}
+		if err != nil {
+			continue
+		}
+		for _, t := range g.targets {
+			if o := got[t]; !o.Found || !(o.Running || o.Restarting) {
+				return true
+			}
+		}
+	}
+	return false
+}

cmd/mesh-host/inactive_test.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/novox/mesh-host/internal/liveness"
+)
+
+// 1957/AC2 and AC4 (unit; ADR 0568 §2, §4): the judge's look acts on a resource that goes down, once per edge. A repeated
+// statement acts nothing, the edge clears when the resource is healthy again, and a resource held, restarting, starting or
+// missing (the other path's) is no edge.
+func testOnlyAResourceNewlyDownIsAnEdge(t *testing.T) {
+	state := func(id, s, reason string, missing bool) liveness.State {
+		return liveness.State{Resource: liveness.Resource{Module: "dnsmasq", ID: id}, State: s, Reason: reason, Missing: missing}
+	}
+	down := state("dnsmasq.unit", liveness.Unhealthy, liveness.ReasonDown, false)
+	steps := []struct {
+		name string
+		st   []liveness.State
+		want []string
+	}{
+		{"down: an edge", []liveness.State{down}, []string{"dnsmasq"}},
+		{"said again while down: none", []liveness.State{down}, nil},
+		{"healthy again: clears", []liveness.State{state("dnsmasq.unit", liveness.Healthy, "", false)}, nil},
+		{"down again: an edge", []liveness.State{down}, []string{"dnsmasq"}},
+		{"held by a window: none", []liveness.State{state("dnsmasq.unit", liveness.Held, "", false)}, nil},
+		{"restarting: none", []liveness.State{state("dnsmasq.unit", liveness.Unhealthy, liveness.ReasonRestarting, false)}, nil},
+		{"starting: none", []liveness.State{state("dnsmasq.unit", liveness.Starting, "", false)}, nil},
+		{"missing: the other path's", []liveness.State{state("dnsmasq.unit", liveness.Unhealthy, liveness.ReasonDown, true)}, nil},
+	}
+	var last map[string]bool
+	for _, s := range steps {
+		var got []string
+		got, last = newlyInactive(last, liveness.Statement{Resources: s.st})
+		if len(got) != len(s.want) || (len(got) == 1 && got[0] != s.want[0]) {
+			t.Fatalf("%s: acted on %v, want %v", s.name, got, s.want)
+		}
+	}
+}
+
+type fakeRuntime struct {
+	units      map[string]liveness.Observed
+	containers map[string]liveness.Observed
+	err        error
+}
+
+func (f fakeRuntime) Containers(_ context.Context, _ []string) (map[string]liveness.Observed, error) {
+	return f.containers, f.err
+}
+func (f fakeRuntime) Units(_ context.Context, _, _ string, _ []string) (map[string]liveness.Observed, error) {
+	return f.units, f.err
+}
+
+// 1957/AC1 (unit; ADR 0568 §1): the machine's facts say a declared-running resource is not active when its unit is stopped
+// or unknown to the manager; everything up, a restart in progress, a read that failed and a container held by a
+// maintenance window are not.
+func testTheMachineShowsAResourceNotActive(t *testing.T) {
+	svc := []byte(`{"id":"dnsmasq.unit","type":"service","unit":"dnsmasq.service","state":"running","module":"dnsmasq"}`)
+	ctr := []byte(`{"id":"dnsmasq.box","type":"container","name":"box","image":"x@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","module":"dnsmasq"}`)
+	up := liveness.Observed{Found: true, Running: true}
+	for _, c := range []struct {
+		name string
+		rt   fakeRuntime
+		held map[string]bool
+		want bool
+	}{
+		{"all up", fakeRuntime{units: map[string]liveness.Observed{"dnsmasq.service": up}, containers: map[string]liveness.Observed{"box": up}}, nil, false},
+		{"unit inactive", fakeRuntime{units: map[string]liveness.Observed{"dnsmasq.service": {Found: true}}, containers: map[string]liveness.Observed{"box": up}}, nil, true},
+		{"unit unknown to the manager", fakeRuntime{containers: map[string]liveness.Observed{"box": up}}, nil, true},
+		{"unit restarting", fakeRuntime{units: map[string]liveness.Observed{"dnsmasq.service": {Found: true, Restarting: true}}, containers: map[string]liveness.Observed{"box": up}}, nil, false},
+		{"container not running", fakeRuntime{units: map[string]liveness.Observed{"dnsmasq.service": up}, containers: map[string]liveness.Observed{"box": {Found: true}}}, nil, true},
+		{"container held by a window", fakeRuntime{units: map[string]liveness.Observed{"dnsmasq.service": up}, containers: map[string]liveness.Observed{"box": {Found: true}}}, map[string]bool{"box": true}, false},
+		{"unread", fakeRuntime{err: errors.New("the service manager could not be read")}, nil, false},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			d, err := declarationOf(delivery.Composed{Node: "n1", Module: "dnsmasq", Resources: []json.RawMessage{svc, ctr}})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if got := anyInactive(context.Background(), c.rt, liveness.LongRunning(d, nil), c.held); got != c.want {
+				t.Fatalf("inactive = %v, want %v", got, c.want)
+			}
+		})
+	}
+}

cmd/mesh-host/main.go

@@ -1046,15 +1046,7 @@ func runLink(ctx context.Context, opts options) error {
 			say(err.Error())
 		}
 		windows := apply.WindowsIn(filepath.Dir(opts.state))
-		j.HeldNow = func(now time.Time) map[string]bool {
-			held := map[string]bool{}
-			for _, w := range windows.Open(now) {
-				for _, name := range w.Holds {
-					held[name] = true
-				}
-			}
-			return held
-		}
+		j.HeldNow = heldBy(windows)
 		judging = j
 		// And which units its service managers say failed, and whose each is (novox/hq issue 315).
 		unitJudge = units.New(units.Exec{Run: apply.ExecRunner, System: builtFor})
@@ -1173,6 +1165,10 @@ func runLink(ctx context.Context, opts options) error {
 			if ag := current.get(); ag != nil {
 				ag.Retry(modules...)
 			}
+		}, func(modules ...string) {
+			if ag := current.get(); ag != nil {
+				ag.Reconcile(modules...)
+			}
 		})
 	}
 	// The node follows its wants: one module at a time from the composition its want names (novox/hq ADR 0539), and the
@@ -1336,8 +1332,8 @@ const (
 // the bus when a state changed, again every minute while one is not healthy, and every five minutes
 // anyway. A statement the bus did not take is said at the next look. It reads; it never acts (ADR 0240
 // rule 6) — but for going on with a take act stopped at its bound (ADR 0328 §7), which runs beside it.
-func judgeWhatRuns(ctx context.Context, j *liveness.Judge, queue *link.Session, say link.Announce, reapply func(modules ...string)) {
-	var missingBefore map[string]bool
+func judgeWhatRuns(ctx context.Context, j *liveness.Judge, queue *link.Session, say link.Announce, reapply, reconcile func(modules ...string)) {
+	var missingBefore, inactiveBefore map[string]bool
 	ticker := time.NewTicker(liveness.LookEvery)
 	defer ticker.Stop()
 	var lastSaid time.Time
@@ -1439,6 +1435,14 @@ func judgeWhatRuns(ctx context.Context, j *liveness.Judge, queue *link.Session,
 				say(fmt.Sprintf("applying %s again: a resource it declares is not on this machine", strings.Join(m, ", ")))
 				reapply(m...)
 			}
+			// A resource its module declares running that is down but still known to the manager is started by the same
+			// apply, once each time it goes down, never on a clock (ADR 0568 §2): the judge reads, the agent acts.
+			var inactive []string
+			inactive, inactiveBefore = newlyInactive(inactiveBefore, st)
+			if len(inactive) > 0 && reconcile != nil {
+				say(fmt.Sprintf("applying %s again: a resource it declares running is not active", strings.Join(inactive, ", ")))
+				reconcile(inactive...)
+			}
 		}
 		if queue.SayHealth(ctx, *withAccesses(withFoundDirectories(withAccounts(withUnits(healthAsReported(st, netSt),
 			unitSt), accountSt), found), accessSt)) {
@@ -2134,6 +2138,32 @@ func newlyMissing(last map[string]bool, st liveness.Statement) ([]string, map[st
 	return missingModules(fresh), now
 }
 
+// newlyInactive are the modules of the resources a statement says down (not running after their grace, but still known to
+// their manager, and not restarting) that the last statement acted on did not, each once, sorted, and the resources down
+// now: a resource still down is not acted on again, and one that came back and went down again is (ADR 0568 §2).
+func newlyInactive(last map[string]bool, st liveness.Statement) ([]string, map[string]bool) {
+	now := map[string]bool{}
+	var fresh liveness.Statement
+	for _, r := range st.Resources {
+		if r.State != liveness.Unhealthy || r.Reason != liveness.ReasonDown || r.Missing {
+			continue
+		}
+		now[r.ID] = true
+		if !last[r.ID] {
+			fresh.Resources = append(fresh.Resources, r)
+		}
+	}
+	return missingModules(withMissing(fresh)), now
+}
+
+// withMissing is the statement with each of its resources marked missing, so missingModules lists their modules.
+func withMissing(st liveness.Statement) liveness.Statement {
+	for i := range st.Resources {
+		st.Resources[i].Missing = true
+	}
+	return st
+}
+
 // recordKeys reads the public halves of the node's keys and the certificate the mesh issued for its internal name from
 // its state directory (novox/hq ADR 0555 §3): what the node's record says of them. A key that cannot be read is left
 // empty, never invented; the certificate is there only after the first declaration that carried it.

cmd/mesh-host/nodeagent.go

@@ -20,6 +20,7 @@ import (
 	"github.com/novox/mesh-host/internal/agent"
 	"github.com/novox/mesh-host/internal/apply"
 	"github.com/novox/mesh-host/internal/link"
+	"github.com/novox/mesh-host/internal/liveness"
 	"github.com/novox/mesh-host/internal/selfcompose"
 	"github.com/novox/mesh-host/internal/store"
 )
@@ -468,6 +469,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
 			// The same kept declaration's filled paths, so a module still running from it keeps its files filled.
 			KeptFilled: func() map[string][]string { _, filled := kept(); return filled },
 			Recorded:   recordedIn(opts.state, say),
+			Inactive:   inactiveIn(&liveness.Exec{Run: apply.ExecRunner, System: builtFor}, heldBy(apply.WindowsIn(filepath.Dir(opts.state)))),
 			Revisions: func(ctx context.Context, key string) ([]agent.Update, error) {
 				got, err := session.KVHistory(ctx, composedBucket, key)
 				if err != nil {

cmd/mesh-host/nodeagent_integration_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/novox/mesh-host/internal/agent"
 	"github.com/novox/mesh-host/internal/link"
 	"github.com/novox/mesh-host/internal/selfcompose"
+	"github.com/novox/mesh-host/internal/store"
 )
 
 // The node agent whole, against a real bus with JetStream started in the test: the wants bucket, the node's composition
@@ -239,6 +240,7 @@ func within(t *testing.T, d time.Duration, what string, ok func() bool) {
 // applied within one second and its have names the fingerprint and the digest; a composition whose digest no want names
 // is not applied.
 func TestNatsAWantAndItsCompositionAreAppliedWithinASecond(t *testing.T) {
+	t.Run("1957: an engine start starts a declared-running unit", testNatsAnEngineStartStartsADeclaredRunningUnitTheHaveSaysRuns)
 	b := newItBed(t)
 	stop := b.start(false)
 	defer stop()
@@ -708,3 +710,71 @@ func h8rest(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, hkey str
 		t.Errorf("a candidate moved what runs: %+v, %q", h, b.file("web"))
 	}
 }
+
+// 1957/AC1 (integration; ADR 0568 §1): a node agent starting on a bus that holds a `running` have for a module at its want,
+// with a service manager (a systemctl on the PATH) that shows the module's unit inactive, starts the unit once and writes
+// the have running; the same start with the unit active applies nothing.
+func testNatsAnEngineStartStartsADeclaredRunningUnitTheHaveSaysRuns(t *testing.T) {
+	for _, active := range []bool{false, true} {
+		b := newItBed(t)
+		bin := t.TempDir()
+		stateFile, calls := filepath.Join(bin, "state"), filepath.Join(bin, "calls")
+		if active {
+			os.WriteFile(stateFile, []byte("active"), 0o644)
+		} else {
+			os.WriteFile(stateFile, []byte("inactive"), 0o644)
+		}
+		script := "#!/bin/sh\necho \"$@\" >> " + calls + "\ncase \"$1\" in\n" +
+			"show) printf 'LoadState=loaded\\nActiveState=%s\\nType=simple\\n' \"$(cat " + stateFile + ")\" ;;\n" +
+			"start) printf active > " + stateFile + " ;;\nis-enabled) echo enabled ;;\nesac\nexit 0\n"
+		if err := os.WriteFile(filepath.Join(bin, "systemctl"), []byte(script), 0o755); err != nil {
+			t.Fatal(err)
+		}
+		t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH"))
+
+		// The engine's own record of the unit, as a previous run left it, and the have it wrote then.
+		res, _ := json.Marshal(map[string]any{"id": "web.unit", "type": "service", "unit": "web-it.service", "state": "running", "module": "web"})
+		var known store.State
+		known.Record(store.Applied{ID: "web.unit", Type: "service", Target: "web-it.service"})
+		if err := store.Save(b.opts.state, known); err != nil {
+			t.Fatal(err)
+		}
+		fp := fpIt("a")
+		c := delivery.Composed{Node: itNode, Module: "web", FP: fp, Resources: []json.RawMessage{res}}
+		signed, _ := delivery.SignComposed(c, b.controll)
+		raw, _ := delivery.EncodeComposed(signed)
+		dig, _ := delivery.ComposedDigestOf(signed)
+		key, _ := delivery.ComposedKey(itNode, "web", fp)
+		if _, err := b.composed.Put(key, raw); err != nil {
+			t.Fatal(err)
+		}
+		b.want("web", fp, dig, "")
+		hkey, _ := delivery.HaveKey(itNode, "web")
+		have, err := delivery.EncodeHave(delivery.Have{FP: fp, Composed: dig, Phase: delivery.PhaseRunning, Health: delivery.HealthUnknown})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := b.haves.Put(hkey, have); err != nil {
+			t.Fatal(err)
+		}
+
+		stop := b.start(false)
+		within(t, 5*time.Second, "the unit was not started and the have not running", func() bool {
+			got, _ := os.ReadFile(stateFile)
+			h := b.have("web")
+			return string(got) == "active" && h.Phase == delivery.PhaseRunning && h.Composed == dig
+		})
+		time.Sleep(300 * time.Millisecond)
+		stop()
+		starts := 0
+		log, _ := os.ReadFile(calls)
+		for _, l := range strings.Split(string(log), "\n") {
+			if strings.HasPrefix(l, "start") {
+				starts++
+			}
+		}
+		if want := map[bool]int{false: 1, true: 0}[active]; starts != want {
+			t.Fatalf("unit active=%v: started %d times, want %d", active, starts, want)
+		}
+	}
+}

cmd/mesh-host/recorded_test.go

@@ -36,6 +36,7 @@ func TestACompositionIsRecordedOnlyWhenEveryResourceIs(t *testing.T) {
 
 // The modules applied again are those with a missing resource, each once; one down but there is not among them.
 func TestTheModulesAppliedAgainAreThoseWithAMissingResource(t *testing.T) {
+	t.Run("1957: the machine shows a resource not active", testTheMachineShowsAResourceNotActive)
 	st := liveness.Statement{Resources: []liveness.State{
 		{Resource: liveness.Resource{Module: "matrix", ID: "matrix.homeserver"}, State: liveness.Unhealthy, Missing: true},
 		{Resource: liveness.Resource{Module: "matrix", ID: "matrix.db"}, State: liveness.Unhealthy, Missing: true},
@@ -51,6 +52,7 @@ func TestTheModulesAppliedAgainAreThoseWithAMissingResource(t *testing.T) {
 // not acted on again, so a resource that cannot be made is not applied (and backed up) at every unrelated change. One
 // that comes back and goes again is newly missing again (the review of issue 1945).
 func TestOnlyANewlyMissingResourceAppliesItsModuleAgain(t *testing.T) {
+	t.Run("1957: a resource newly down is an edge", testOnlyAResourceNewlyDownIsAnEdge)
 	missing := func(ids ...string) liveness.Statement {
 		var st liveness.Statement
 		for _, id := range ids {