Z9 zfs-apply-run: the apply holds the lock, waits on a scrub or an unimported pool, and converges the node's id, ARC and pool

Builds: design 42 zfs-owns-zfs §15, §18, A-8, A-16, A-29

modules/zfs/cmd/zfs-tools/applied.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"time"
 )
@@ -41,6 +42,9 @@ const (
 	OutcomeRefused = "refused"
 	OutcomeWaiting = "waiting"
 	OutcomeFailed  = "failed"
+	// OutcomeAwaitingLoad is a kernel module parameter set in the options file whose live value differs: it
+	// takes effect at the next load of the kernel module (§18, §29).
+	OutcomeAwaitingLoad = "awaiting-load"
 )
 
 // StateStore is the module's states on the bus, as the process or the tools bundle reaches them.
@@ -49,6 +53,8 @@ type StateStore interface {
 	Delete(state, key string) error
 	// Get answers false when the key is not there.
 	Get(state, key string) (json.RawMessage, bool, error)
+	// Keys are the keys of a state.
+	Keys(state string) ([]string, error)
 }
 
 // AppliedEntry is one key of the state `applied`.
@@ -63,21 +69,47 @@ type AppliedEntry struct {
 	Since   time.Time `json:"since,omitzero"`
 }
 
+// notSaid is a change, refusal or wait the bus did not take: the machine is as the apply left it, but the
+// event or the key is missing. It is a warning, never a failed command (§18: with the bus down the apply
+// converges from the config file it last got).
+type notSaid struct{ err error }
+
+func (e notSaid) Error() string { return e.err.Error() }
+func (e notSaid) Unwrap() error { return e.err }
+
 // say publishes one event of the module.
 func (m *Machine) say(kind string, body map[string]any) error {
 	if m.Emit == nil {
-		return fmt.Errorf("this process publishes no events: %s was not said", kind)
+		return notSaid{fmt.Errorf("this process publishes no events: %s was not said", kind)}
 	}
-	return m.Emit(kind, body)
+	if err := m.Emit(kind, body); err != nil {
+		return notSaid{err}
+	}
+	return nil
+}
+
+// failure records one failed step of the apply: a failed command is kept under its key and answered as the
+// apply's failure; a change the bus did not take is only a warning.
+func (m *Machine) failure(failed *[]error, key string, err error) {
+	var ns notSaid
+	if errors.As(err, &ns) {
+		m.Warnings = append(m.Warnings, fmt.Sprintf("%s: %v", key, err))
+		return
+	}
+	*failed = append(*failed, fmt.Errorf("%s: %v", key, err))
+	_ = m.CommandFailed(key, err)
 }
 
 // keep writes a key of the state `applied`.
 func (m *Machine) keep(key string, e AppliedEntry) error {
 	if m.States == nil {
-		return fmt.Errorf("this process keeps no state: %s was not kept", key)
+		return notSaid{fmt.Errorf("this process keeps no state: %s was not kept", key)}
 	}
 	e.At = m.Now().UTC()
-	return m.States.Put(StateApplied, key, e)
+	if err := m.States.Put(StateApplied, key, e); err != nil {
+		return notSaid{err}
+	}
+	return nil
 }
 
 // Changed says a change as `config.applied` {target, key, before, after, read-back} and keeps it. The event

modules/zfs/cmd/zfs-tools/applied_test.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"encoding/json"
+	"sort"
 	"strings"
 	"testing"
 	"time"
@@ -15,6 +16,16 @@ func (s memStates) Put(state, key string, v any) error {
 	return nil
 }
 func (s memStates) Delete(state, key string) error { delete(s, state+"/"+key); return nil }
+func (s memStates) Keys(state string) ([]string, error) {
+	var out []string
+	for k := range s {
+		if rest, ok := strings.CutPrefix(k, state+"/"); ok {
+			out = append(out, rest)
+		}
+	}
+	sort.Strings(out)
+	return out, nil
+}
 func (s memStates) Get(state, key string) (json.RawMessage, bool, error) {
 	b, ok := s[state+"/"+key]
 	return b, ok, nil

modules/zfs/cmd/zfs-tools/apply.go

@@ -0,0 +1,401 @@
+package main
+
+import (
+	"context"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"os"
+	"regexp"
+	"strconv"
+	"strings"
+	"syscall"
+)
+
+// The apply: who, when, and what it converges (design 42 zfs-owns-zfs §15, §18).
+//
+// `zfs-tools apply` runs as root, holds an exclusive lock on ApplyLock (a second start waits for it), reads
+// the config file the node-engine wrote, and converges the node's ZFS id, the ARC's live parameters, the
+// pool and dataset properties and the cache and log devices. It converges nothing while a scrub or resilver
+// runs on the pool or before the pool is imported: it says `config.waiting` and ends, and the event follower
+// starts it again on the event that ends the wait (§27). It never runs on a timer. It exits 0 on a refusal or
+// a wait, and non-zero only when a command it ran failed (A-8), so the node-engine gates the module on a real
+// failure.
+
+// ApplyLock is the lock the apply, the scheduled run and the hand acts take.
+const ApplyLock = "/run/zfs-tools/apply.lock"
+
+// The files the apply reads and writes.
+const (
+	HostIDFile    = "/etc/hostid"
+	ParametersDir = "/sys/module/zfs/parameters/"
+	MemInfoFile   = "/proc/meminfo"
+)
+
+// ZdbConfigArgv reads a pool's configuration, where its label's hostid is.
+func ZdbConfigArgv(pool string) []string { return []string{"-C", pool} }
+
+// ZgenhostidArgv writes /etc/hostid with ZFS's own writer; the value is one the parse validated (§11).
+func ZgenhostidArgv(value string) []string { return []string{"-f", value} }
+
+// FileLock takes an exclusive lock on a file, waiting for it or answering at once.
+func FileLock(path string, wait bool) (func(), bool, error) {
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
+	if err != nil {
+		return nil, false, err
+	}
+	how := syscall.LOCK_EX
+	if !wait {
+		how |= syscall.LOCK_NB
+	}
+	for {
+		err = syscall.Flock(int(f.Fd()), how)
+		if err != syscall.EINTR {
+			break
+		}
+	}
+	if err != nil {
+		f.Close()
+		if err == syscall.EWOULDBLOCK {
+			return nil, false, nil
+		}
+		return nil, false, err
+	}
+	return func() { f.Close() }, true, nil
+}
+
+func writeLive(path string, data []byte) error { return os.WriteFile(path, data, 0o644) }
+
+// memMiB is the node's memory in MiB from /proc/meminfo, 0 when it cannot be read (A-29).
+func (m *Machine) memMiB() uint64 {
+	b, err := m.ReadFile(MemInfoFile)
+	if err != nil {
+		return 0
+	}
+	for _, l := range strings.Split(string(b), "\n") {
+		if rest, ok := strings.CutPrefix(l, "MemTotal:"); ok {
+			f := strings.Fields(rest)
+			if len(f) > 0 {
+				if kb, err := strconv.ParseUint(f[0], 10, 64); err == nil {
+					return kb / 1024
+				}
+			}
+		}
+	}
+	return 0
+}
+
+// Managed says whether the config value a state key belongs to (its words before the first dot) is set.
+func (c *Config) Managed(key string) bool {
+	k, _, _ := strings.Cut(key, ".")
+	switch k {
+	case "pool":
+		return c.Pool != ""
+	case "scrub-cadence":
+		return c.ScrubCadence != ""
+	case "scrub-day":
+		return c.ScrubDay != ""
+	case "scrub-time":
+		return c.ScrubTime != ""
+	case "arc-max":
+		return c.ArcMax != nil
+	case "arc-min":
+		return c.ArcMin != nil
+	case "module-options":
+		return len(c.ModuleOptions) > 0
+	case "hostid":
+		return c.HostID != ""
+	case "zed":
+		return c.Zed != ""
+	case "zed-handlers":
+		return len(c.ZedHandlers) > 0
+	case "trim-schedule":
+		return c.TrimSchedule != ""
+	case "trim-day":
+		return c.TrimDay != ""
+	case "trim-time":
+		return c.TrimTime != ""
+	case "cache-devices":
+		return len(c.CacheDevices) > 0
+	case "log-devices":
+		return len(c.LogDevices) > 0
+	case "log-mirror":
+		return c.LogMirror != nil
+	case "pool-properties":
+		return len(c.PoolProperties) > 0
+	case "dataset-properties":
+		return len(c.DatasetProperties) > 0
+	case "snapshots":
+		return c.Snapshots != nil
+	}
+	return false
+}
+
+// scanBusy says what runs on the pool that holds the apply back, "" when nothing does.
+func scanBusy(scans []Scan) string {
+	for _, s := range scans {
+		if (s.Kind == "scrub" || s.Kind == "resilver") && (s.State == "running" || s.State == "paused") {
+			return "a " + s.Kind + " is running on the pool"
+		}
+	}
+	return ""
+}
+
+// Apply converges the node to its config file and answers a line of what it did. Its error is a failed
+// command, never a refusal or a wait.
+func (m *Machine) Apply(ctx context.Context) (string, error) {
+	release, _, err := m.Lock(ApplyLock, true)
+	if err != nil {
+		return "", fmt.Errorf("the apply's lock %s: %v", ApplyLock, err)
+	}
+	defer release()
+	text, err := m.ReadFile(ConfigFile)
+	if err != nil {
+		return "", fmt.Errorf("the config file %s cannot be read: %v", ConfigFile, err)
+	}
+	c, refusals := ParseConfig(text, m.memMiB())
+	if len(refusals) > 0 {
+		var failed []error
+		for _, r := range refusals {
+			if err := m.Refused(r); err != nil {
+				m.failure(&failed, r.Key, err)
+			}
+		}
+		return fmt.Sprintf("%d values refused; nothing converged", len(refusals)), errors.Join(failed...)
+	}
+	m.forget(c)
+	var failed []error
+	// The pool must be imported and idle; until then nothing converges.
+	if c.Pool != "" {
+		wait, err := m.poolWait(ctx, c.Pool)
+		if err != nil {
+			m.failure(&failed, "pool", err)
+			return "the pool could not be read", errors.Join(failed...)
+		}
+		if wait != "" {
+			if err := m.Waiting("pool", wait); err != nil {
+				m.failure(&failed, "pool", err)
+			}
+			return "waiting for " + wait, errors.Join(failed...)
+		}
+		_ = m.NotManaged("pool")
+	}
+	if c.HostID != "" && c.Pool == "" {
+		_ = m.Refused(Refusal{Key: "hostid", Value: c.HostID, Reason: "needs the pool set, to read its label"})
+	} else if c.HostID != "" {
+		m.ApplyHostID(ctx, c, &failed)
+	}
+	m.ApplyARC(c, &failed)
+	if c.Pool != "" {
+		if err := m.ApplyProperties(ctx, c); err != nil {
+			failed = append(failed, err)
+		}
+		if err := m.ApplyDevices(ctx, c); err != nil {
+			failed = append(failed, err)
+		}
+	}
+	return "applied", errors.Join(failed...)
+}
+
+// forget deletes the keys of the state `applied` that are a refusal or a wait of a value, or a value now not
+// managed: what is still wrong is said again by this apply, and healthy is a key that is not there.
+func (m *Machine) forget(c *Config) {
+	if m.States == nil {
+		return
+	}
+	keys, err := m.States.Keys(StateApplied)
+	if err != nil {
+		return
+	}
+	for _, k := range keys {
+		if !c.Managed(k) {
+			_ = m.States.Delete(StateApplied, k)
+			continue
+		}
+		raw, ok, err := m.States.Get(StateApplied, k)
+		if err != nil || !ok {
+			continue
+		}
+		if strings.Contains(string(raw), `"outcome":"`+OutcomeRefused+`"`) || strings.Contains(string(raw), `"outcome":"`+OutcomeFailed+`"`) {
+			_ = m.States.Delete(StateApplied, k)
+		}
+	}
+}
+
+// poolWait is what the apply waits on, "" when the pool is imported and idle.
+func (m *Machine) poolWait(ctx context.Context, pool string) (string, error) {
+	pools, err := m.Pools(ctx)
+	if err != nil {
+		return "", err
+	}
+	found := false
+	for _, p := range pools {
+		found = found || p.Name == pool
+	}
+	if !found {
+		return "the pool " + pool + " to be imported", nil
+	}
+	out, err := m.ask(ctx, "zpool", StatusArgv()...)
+	if err != nil {
+		return "", err
+	}
+	ps, err := ParseStatus(out)
+	if err != nil {
+		return "", err
+	}
+	for _, p := range ps {
+		if p.Name == pool {
+			if busy := scanBusy(p.Scans); busy != "" {
+				return "the scrub or resilver of the pool " + pool + " to end (" + busy + ")", nil
+			}
+		}
+	}
+	return "", nil
+}
+
+var labelHostID = regexp.MustCompile(`(?m)^\s*hostid:\s*(\d+)`)
+
+func nativeHostID(b []byte) (uint32, bool) {
+	if len(b) != 4 {
+		return 0, false
+	}
+	return binary.NativeEndian.Uint32(b), true
+}
+
+// ApplyHostID converges /etc/hostid to the set value, only when the pool's label is 0 or already that value
+// (§15), and reads the file back.
+func (m *Machine) ApplyHostID(ctx context.Context, c *Config, failed *[]error) {
+	const key = "hostid"
+	refuse := func(why string) {
+		if err := m.Refused(Refusal{Key: key, Value: c.HostID, Reason: why}); err != nil {
+			m.failure(failed, key, err)
+		}
+	}
+	n, _ := strconv.ParseUint(c.HostID, 16, 32)
+	want := uint32(n)
+	current := func() string {
+		b, err := m.ReadFile(HostIDFile)
+		if err != nil {
+			return "not there"
+		}
+		if v, ok := nativeHostID(b); ok {
+			return fmt.Sprintf("%08x", v)
+		}
+		return "not four bytes"
+	}
+	r, err := m.AsRoot(ctx, "zdb", ZdbConfigArgv(c.Pool)...)
+	if err != nil {
+		refuse("the pool's label cannot be read: " + err.Error())
+		return
+	}
+	lm := labelHostID.FindStringSubmatch(r.Out)
+	if lm == nil {
+		refuse("the pool's label cannot be read: zdb -C shows no hostid")
+		return
+	}
+	label, _ := strconv.ParseUint(lm[1], 10, 32)
+	if label != 0 && uint32(label) != want {
+		refuse(fmt.Sprintf("the pool's import would need -f: its label holds the hostid %08x, not %s", label, c.HostID))
+		return
+	}
+	before := current()
+	if before != c.HostID {
+		if _, err := m.AsRoot(ctx, "zgenhostid", ZgenhostidArgv(c.HostID)...); err != nil {
+			m.failure(failed, key, err)
+			return
+		}
+	}
+	back := current()
+	if back != c.HostID {
+		refuse(fmt.Sprintf("%s holds %s after the write, not the set %s: another writer changed it", HostIDFile, back, c.HostID))
+		return
+	}
+	if before == back {
+		if err := m.keepAsSet(key, back); err != nil {
+			m.failure(failed, key, err)
+		}
+		return
+	}
+	if err := m.Changed("file", key, before, c.HostID, back); err != nil {
+		m.failure(failed, key, err)
+	}
+}
+
+// ApplyARC writes arc-max and arc-min live and reads both back; a module option whose live value differs is
+// a warning that it takes effect at the next load of the kernel module, never a failure.
+func (m *Machine) ApplyARC(c *Config, failed *[]error) {
+	params := []struct {
+		key, file string
+		mib       *uint64
+	}{{"arc-max", "zfs_arc_max", c.ArcMax}, {"arc-min", "zfs_arc_min", c.ArcMin}}
+	live := func(file string) (string, error) {
+		b, err := m.ReadFile(ParametersDir + file)
+		return strings.TrimSpace(string(b)), err
+	}
+	before := map[string]string{}
+	for _, p := range params {
+		if p.mib == nil {
+			continue
+		}
+		v, err := live(p.file)
+		if err != nil {
+			m.failure(failed, p.key, fmt.Errorf("%s cannot be read: %v", ParametersDir+p.file, err))
+			continue
+		}
+		before[p.key] = v
+	}
+	// arc-min may not pass arc-max while ZFS holds them, so a second pass settles an order that failed.
+	for pass := 0; pass < 2; pass++ {
+		for _, p := range params {
+			if p.mib == nil || before[p.key] == "" {
+				continue
+			}
+			want := strconv.FormatUint(*p.mib<<20, 10)
+			if v, _ := live(p.file); v != want {
+				if err := m.WriteFile(ParametersDir+p.file, []byte(want)); err != nil && pass == 1 {
+					m.failure(failed, p.key, err)
+				}
+			}
+		}
+	}
+	for _, p := range params {
+		if p.mib == nil || before[p.key] == "" {
+			continue
+		}
+		want := strconv.FormatUint(*p.mib<<20, 10)
+		back, _ := live(p.file)
+		var err error
+		switch {
+		case back != want:
+			err = fmt.Errorf("%s reads %q after writing %q", ParametersDir+p.file, back, want)
+			m.failure(failed, p.key, err)
+		case before[p.key] == want:
+			err = m.keepAsSet(p.key, back)
+		default:
+			err = m.Changed("file", p.key, before[p.key], want, back)
+		}
+		if err != nil && back == want {
+			m.failure(failed, p.key, err)
+		}
+	}
+	for _, o := range c.ModuleOptions {
+		key := "module-options." + o.Name
+		v, err := live(o.Name)
+		switch {
+		case err == nil && v == o.Value:
+			if err := m.keepAsSet(key, v); err != nil {
+				m.failure(failed, key, err)
+			}
+		default:
+			if err == nil {
+				m.Warnings = append(m.Warnings, fmt.Sprintf("%s: live %s, set %s: takes effect at the next load of the kernel module", o.Name, v, o.Value))
+			} else {
+				m.Warnings = append(m.Warnings, fmt.Sprintf("%s: not a live parameter here: takes effect at the next load of the kernel module", o.Name))
+			}
+			if err := m.keep(key, AppliedEntry{Before: v, After: o.Value, Outcome: OutcomeAwaitingLoad,
+				Why: "takes effect at the next load of the kernel module"}); err != nil {
+				m.failure(failed, key, err)
+			}
+		}
+	}
+}

modules/zfs/cmd/zfs-tools/apply_devices.go

@@ -145,10 +145,8 @@ func (m *Machine) devicesOf(ctx context.Context, pool string) (*poolDevices, err
 	case real.Root != nil && deviceErrors(real.Root) > 0:
 		pd.Healthy, pd.Why = false, "a device of the pool has read, write or checksum errors"
 	}
-	for _, s := range real.Scans {
-		if (s.Kind == "scrub" || s.Kind == "resilver") && (s.State == "running" || s.State == "paused") {
-			pd.Healthy, pd.Why = false, "a "+s.Kind+" is running on the pool"
-		}
+	if busy := scanBusy(real.Scans); busy != "" {
+		pd.Healthy, pd.Why = false, busy
 	}
 	for _, role := range []string{"cache", "log"} {
 		hs, rs := sectionOf(held, role), sectionOf(real, role)
@@ -302,10 +300,7 @@ func (m *Machine) ApplyDevices(ctx context.Context, c *Config) error {
 		return nil
 	}
 	var failed []error
-	fail := func(key string, err error) {
-		failed = append(failed, fmt.Errorf("%s: %v", key, err))
-		_ = m.CommandFailed(key, err)
-	}
+	fail := func(key string, err error) { m.failure(&failed, key, err) }
 	others, err := m.otherPoolDevices(ctx, c.Pool)
 	if err != nil {
 		fail("cache-devices", err)

modules/zfs/cmd/zfs-tools/apply_properties.go

@@ -75,10 +75,7 @@ func sameValue(property, got, want string) bool {
 // one error; a property whose read-back differs from what was set is a failure too.
 func (m *Machine) ApplyProperties(ctx context.Context, c *Config) error {
 	var failed []error
-	fail := func(key string, err error) {
-		failed = append(failed, fmt.Errorf("%s: %v", key, err))
-		_ = m.CommandFailed(key, err)
-	}
+	fail := func(key string, err error) { m.failure(&failed, key, err) }
 	for _, p := range sortedKeys(c.PoolProperties) {
 		key := "pool-properties." + p
 		want := c.PoolProperties[p]

modules/zfs/cmd/zfs-tools/apply_test.go

@@ -0,0 +1,205 @@
+package main
+
+import (
+	"context"
+	"encoding/binary"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func hostidFile(v uint32) string {
+	b := make([]byte, 4)
+	binary.NativeEndian.PutUint32(b, v)
+	return string(b)
+}
+
+// applyFake is a machine whose pool tank is imported and idle, with the memory and the ARC files of a node.
+func applyFake(t *testing.T, conf string) *fake {
+	f := newFake(t)
+	f.uid = 0
+	f.files[ConfigFile] = conf
+	f.files[MemInfoFile] = "MemTotal:       16384000 kB\n"
+	f.on(line("zpool", PoolsArgv()), fixture(t, "zpool-list.txt"))
+	f.on(line("zpool", StatusArgv()), fixture(t, "zpool-status.txt"))
+	f.files[ParametersDir+"zfs_arc_max"] = "0\n"
+	f.files[ParametersDir+"zfs_arc_min"] = "0\n"
+	return f
+}
+
+func hostidCase(f *fake, label, file string) {
+	f.on("zdb -C tank", "MOS Configuration:\n    version: 5000\n    name: 'tank'\n    hostid: "+label+"\n")
+	if file != "" {
+		f.files[HostIDFile] = file
+	}
+}
+
+// §15, §31: unset: no zgenhostid; a label of another id or one that cannot be read: refused; 0 or equal: one
+// run; a file holding another value after the write: refused; each refusal unhealthy and healthy again once
+// the label and the file agree or hostid is unset.
+func TestTheHostIDIsConvergedOnlyWhereTheLabelAllowsIt(t *testing.T) {
+	const want = "00bab10c" // 12235020
+	cases := []struct {
+		name, conf, label, file string
+		zdbErr                  bool
+		writes                  bool
+		event                   string
+		outcome                 string
+	}{
+		{name: "unset", conf: `{"pool":"tank"}`},
+		{name: "label of another id", conf: `{"pool":"tank","hostid":"` + want + `"}`, label: "12345", event: "config.refused", outcome: "refused"},
+		{name: "label cannot be read", conf: `{"pool":"tank","hostid":"` + want + `"}`, zdbErr: true, event: "config.refused", outcome: "refused"},
+		{name: "label 0", conf: `{"pool":"tank","hostid":"` + want + `"}`, label: "0", writes: true, event: "config.applied", outcome: "applied"},
+		{name: "label equal", conf: `{"pool":"tank","hostid":"` + want + `"}`, label: "12235020", writes: true, event: "config.applied", outcome: "applied"},
+		{name: "file already equal", conf: `{"pool":"tank","hostid":"` + want + `"}`, label: "12235020", file: hostidFile(12235020), outcome: "applied"},
+		{name: "another writer's value after the write", conf: `{"pool":"tank","hostid":"` + want + `"}`, label: "0", file: hostidFile(7), writes: true, event: "config.refused", outcome: "refused"},
+	}
+	for _, c := range cases {
+		f := applyFake(t, c.conf)
+		if c.zdbErr {
+			f.answers["zdb -C tank"] = Ran{Code: 1, Err: "cannot open"}
+		} else if c.label != "" {
+			hostidCase(f, c.label, c.file)
+		}
+		if c.writes {
+			f.on("zgenhostid -f "+want, "")
+			f.after["zgenhostid -f "+want] = func(f *fake) {
+				if c.name != "another writer's value after the write" {
+					f.files[HostIDFile] = hostidFile(12235020)
+				}
+			}
+		}
+		m := f.machine()
+		st := memStates{}
+		m.States = st
+		if _, err := m.Apply(context.Background()); err != nil {
+			t.Errorf("%s: %v", c.name, err)
+		}
+		if n := len(ran(f, "zgenhostid")); (n == 1) != c.writes || n > 1 {
+			t.Errorf("%s: zgenhostid ran %d times", c.name, n)
+		}
+		if c.event == "" && len(f.emitted) != 0 || c.event != "" && (len(f.emitted) != 1 || f.emitted[0]["_type"] != c.event) {
+			t.Errorf("%s: events %v, want %s", c.name, f.emitted, c.event)
+		}
+		if raw, ok := st["applied/hostid"]; c.outcome != "" && (!ok || !strings.Contains(string(raw), `"outcome":"`+c.outcome+`"`)) {
+			t.Errorf("%s: state %v", c.name, st)
+		}
+		if c.outcome == "refused" {
+			// Healthy again once hostid is unset.
+			f.files[ConfigFile] = `{"pool":"tank"}`
+			if _, err := m.Apply(context.Background()); err != nil {
+				t.Error(err)
+			}
+			if _, ok := st["applied/hostid"]; ok {
+				t.Errorf("%s: the refusal stayed after hostid was unset", c.name)
+			}
+		}
+	}
+}
+
+// §18: the lock is waited for; a scrub, resilver or unimported pool is a wait and the apply ends with exit 0;
+// exit non-zero only for a failed command; ARC values are written live and read back, a differing module
+// option is a warning; with the bus down it converges from the config file.
+func TestTheApplyWaitsConvergesAndExitsAsADR0679Says(t *testing.T) {
+	t.Run("lock waited for, ARC written and read back", func(t *testing.T) {
+		f := applyFake(t, `{"pool":"tank","arc-max":4096,"arc-min":512}`)
+		m := f.machine()
+		m.States = memStates{}
+		if _, err := m.Apply(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+		if len(f.locks) != 1 || !f.locks[0] {
+			t.Errorf("locks %v: the apply must wait for the lock", f.locks)
+		}
+		if f.files[ParametersDir+"zfs_arc_max"] != "4294967296" || f.files[ParametersDir+"zfs_arc_min"] != "536870912" {
+			t.Errorf("ARC %q %q", f.files[ParametersDir+"zfs_arc_max"], f.files[ParametersDir+"zfs_arc_min"])
+		}
+		if len(f.emitted) != 2 || f.emitted[0]["read-back"] != "4294967296" {
+			t.Errorf("events %v", f.emitted)
+		}
+	})
+	t.Run("a scrub running is a wait", func(t *testing.T) {
+		f := applyFake(t, `{"pool":"tank","arc-max":4096}`)
+		f.on(line("zpool", StatusArgv()), fixture(t, "zpool-status-scrubbing.txt"))
+		m := f.machine()
+		m.States = memStates{}
+		if _, err := m.Apply(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+		if len(f.emitted) != 1 || f.emitted[0]["_type"] != "config.waiting" || f.files[ParametersDir+"zfs_arc_max"] != "0\n" {
+			t.Errorf("events %v", f.emitted)
+		}
+	})
+	t.Run("an unimported pool is a wait", func(t *testing.T) {
+		f := applyFake(t, `{"pool":"nopool","arc-max":4096}`)
+		m := f.machine()
+		m.States = memStates{}
+		if _, err := m.Apply(context.Background()); err != nil || len(f.emitted) != 1 || f.emitted[0]["_type"] != "config.waiting" {
+			t.Errorf("%v %v", err, f.emitted)
+		}
+	})
+	t.Run("a refused value refuses the whole file, exit 0", func(t *testing.T) {
+		f := applyFake(t, `{"pool":"tank","arc-max":4096,"module-options":["bogus=1"]}`)
+		m := f.machine()
+		m.States = memStates{}
+		if _, err := m.Apply(context.Background()); err != nil || len(f.emitted) != 1 || f.emitted[0]["_type"] != "config.refused" ||
+			f.files[ParametersDir+"zfs_arc_max"] != "0\n" {
+			t.Errorf("%v %v", err, f.emitted)
+		}
+	})
+	t.Run("a failed command is the only non-zero exit", func(t *testing.T) {
+		f := applyFake(t, `{"pool":"tank","pool-properties":{"autotrim":"on"}}`)
+		f.on("zpool get -Hp -o value autotrim -- tank", "off\n")
+		f.answers["zpool set autotrim=on -- tank"] = Ran{Code: 1, Err: "cannot set"}
+		m := f.machine()
+		m.States = memStates{}
+		if _, err := m.Apply(context.Background()); err == nil {
+			t.Error("a failed zpool set exited 0")
+		}
+	})
+	t.Run("a differing module option is a warning, never a failure", func(t *testing.T) {
+		f := applyFake(t, `{"pool":"tank","module-options":["zfs_txg_timeout=10"]}`)
+		f.files[ParametersDir+"zfs_txg_timeout"] = "5\n"
+		m := f.machine()
+		st := memStates{}
+		m.States = st
+		if _, err := m.Apply(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+		if len(m.Warnings) != 1 || !strings.Contains(m.Warnings[0], "next load of the kernel module") ||
+			!strings.Contains(string(st["applied/module-options.zfs_txg_timeout"]), OutcomeAwaitingLoad) || len(f.emitted) != 0 {
+			t.Errorf("%v %v %v", m.Warnings, st, f.emitted)
+		}
+	})
+	t.Run("with the bus down it converges from the config file", func(t *testing.T) {
+		f := applyFake(t, `{"pool":"tank","arc-max":4096}`)
+		m := f.machine()
+		m.Emit, m.States = nil, nil
+		if _, err := m.Apply(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+		if f.files[ParametersDir+"zfs_arc_max"] != "4294967296" || len(m.Warnings) == 0 {
+			t.Errorf("ARC %q, warnings %v", f.files[ParametersDir+"zfs_arc_max"], m.Warnings)
+		}
+	})
+}
+
+// §18: FileLock holds a second taker out, and a second one that waits gets it once the first lets go.
+func TestFileLockHoldsASecondTakerOut(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "apply.lock")
+	release, got, err := FileLock(path, false)
+	if err != nil || !got {
+		t.Fatal(err, got)
+	}
+	if _, got2, err := FileLock(path, false); err != nil || got2 {
+		t.Errorf("second taker: %v %v", got2, err)
+	}
+	release()
+	rel2, got3, err := FileLock(path, false)
+	if err != nil || !got3 {
+		t.Errorf("after release: %v %v", got3, err)
+	}
+	rel2()
+	_ = os.Remove(path)
+}

modules/zfs/cmd/zfs-tools/bus.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/process"
+)
+
+// busStates is the module's states as this process reaches them (its bus account's `state`).
+type busStates struct{ c *process.Conn }
+
+func (b busStates) Put(state, key string, v any) error {
+	_, err := b.c.State(state).Put(key, v)
+	return err
+}
+func (b busStates) Delete(state, key string) error      { return b.c.State(state).Delete(key) }
+func (b busStates) Keys(state string) ([]string, error) { return b.c.State(state).Keys() }
+func (b busStates) Get(state, key string) (json.RawMessage, bool, error) {
+	e, err := b.c.State(state).Get(key)
+	if err != nil {
+		return nil, false, err
+	}
+	if e == nil {
+		return nil, false, nil
+	}
+	return e.Value, true, nil
+}
+
+// JoinBus gives the machine this process's events and states, as its own bus account says. Without the
+// account or the bus the machine keeps none, and what it changes is said as a warning (§18: with the bus down
+// the apply converges from the config file it last got). It answers how to leave the bus.
+func (m *Machine) JoinBus() func() {
+	acct, err := process.ReadBrokerAccount(os.Getenv)
+	if err != nil {
+		m.Warnings = append(m.Warnings, err.Error())
+		return func() {}
+	}
+	c, err := process.Connect(acct)
+	if err != nil {
+		m.Warnings = append(m.Warnings, err.Error())
+		return func() {}
+	}
+	m.Emit = func(kind string, body any) error { return c.Emit(kind, body, stdio.EmitOptions{}) }
+	m.States = busStates{c}
+	return c.Close
+}

modules/zfs/cmd/zfs-tools/helpers_test.go

@@ -33,6 +33,7 @@ type fake struct {
 	emitErr error
 	files   map[string]string
 	links   map[string]string
+	locks   []bool // the wait of each lock taken
 	now     time.Time
 	uid     int
 	errors  *ErrorMemory
@@ -81,6 +82,11 @@ func (f *fake) machine() *Machine {
 			}
 			return "", os.ErrNotExist
 		},
+		WriteFile: func(p string, b []byte) error { f.files[p] = string(b); return nil },
+		Lock: func(path string, wait bool) (func(), bool, error) {
+			f.locks = append(f.locks, wait)
+			return func() {}, true, nil
+		},
 		Errors: f.errors,
 		Emit: func(eventType string, body any) error {
 			if f.emitErr != nil {

modules/zfs/cmd/zfs-tools/machine.go

@@ -112,6 +112,14 @@ type Machine struct {
 	Now func() time.Time
 	// ReadFile reads a kernel file (the ARC's counters).
 	ReadFile func(path string) ([]byte, error)
+	// WriteFile writes a file as root (the ARC's live parameters).
+	WriteFile func(path string, data []byte) error
+	// Lock takes an exclusive lock on a file: waiting for it, or answering at once whether it was got. The
+	// release gives it up.
+	Lock func(path string, wait bool) (release func(), got bool, err error)
+	// Warnings are what the apply saw that is no failure (a change the bus did not take, a parameter awaiting
+	// a load).
+	Warnings []string
 	// Resolve follows a link to its kernel path (a declared device's name under /dev/disk/by-id).
 	Resolve func(path string) (string, error)
 	// Emit publishes an event as this module (the record of an act); nil where nothing can be published.
@@ -125,7 +133,7 @@ type Machine struct {
 
 // Here is the machine this process runs on.
 func Here() *Machine {
-	return &Machine{Run: ExecRunner, UID: os.Getuid(), Now: time.Now, ReadFile: os.ReadFile, Resolve: filepath.EvalSymlinks, Errors: LoadErrorMemory(ErrorsFile)}
+	return &Machine{Run: ExecRunner, UID: os.Getuid(), Now: time.Now, ReadFile: os.ReadFile, Resolve: filepath.EvalSymlinks, WriteFile: writeLive, Lock: FileLock, Errors: LoadErrorMemory(ErrorsFile)}
 }
 
 // NotApplicable is the answer of a machine without ZFS: the module does not apply here.

modules/zfs/cmd/zfs-tools/main.go

@@ -34,6 +34,19 @@ func main() {
 				os.Exit(1)
 			}
 			fmt.Println(said)
+		case "apply":
+			m := Here()
+			closeBus := m.JoinBus()
+			said, err := m.Apply(ctx)
+			closeBus()
+			for _, w := range m.Warnings {
+				fmt.Fprintln(os.Stderr, "warning:", w)
+			}
+			if err != nil {
+				fmt.Fprintln(os.Stderr, err)
+				os.Exit(1)
+			}
+			fmt.Println(said)
 		case "events":
 			if err := Follower(ctx, os.Stdout); err != nil {
 				fmt.Fprintln(os.Stderr, err)
@@ -41,7 +54,7 @@ func main() {
 			}
 		default:
 			fmt.Fprintf(os.Stderr, "zfs-tools: %q is not a way to run; with no argument it serves its tools, with watch it "+
-				"watches the pools, with events it follows ZFS's events, with scrub-run it runs the scheduled scrub\n", os.Args[1])
+				"watches the pools, with events it follows ZFS's events, with scrub-run it runs the scheduled scrub, with apply it converges the node to its config file\n", os.Args[1])
 			os.Exit(2)
 		}
 		return

modules/zfs/cmd/zfs-tools/manifest_test.go

@@ -222,6 +222,7 @@ func TestTheBundleRunsOnlyReadingVerbsSnapshotAndScrub(t *testing.T) {
 		"DatasetSetArgv": {"zfs", "set"}, "DatasetInheritArgv": {"zfs", "inherit"},
 		"DeviceStatusArgv": {"zpool", "status"}, "AllRealStatusArgv": {"zpool", "status"}, "RemoveArgv": {"zpool", "remove"},
 		"AddArgv": {"zpool", "add"}, "WipefsArgv": {"wipefs", "--no-act"}, "ZdbLabelArgv": {"zdb", "-l"},
+		"ZdbConfigArgv": {"zdb", "-C"}, "ZgenhostidArgv": {"zgenhostid", "-f"},
 	}
 	verbs := map[string][]string{"DatasetsArgv": DatasetsArgv(), "SnapshotsArgv": SnapshotsArgv(), "PoolsArgv": PoolsArgv(),
 		"StatusArgv": StatusArgv(), "IOArgv": IOArgv(3), "EventsArgv": EventsArgv(), "HistoryArgv": HistoryArgv("tank"),
@@ -232,10 +233,11 @@ func TestTheBundleRunsOnlyReadingVerbsSnapshotAndScrub(t *testing.T) {
 		"DatasetInheritArgv": DatasetInheritArgv("tank", "atime"),
 		"DeviceStatusArgv":   DeviceStatusArgv("tank", true), "AllRealStatusArgv": AllRealStatusArgv(),
 		"RemoveArgv": RemoveArgv("tank", "x"), "AddArgv": AddArgv("tank", "cache", false, "/dev/x"),
-		"WipefsArgv": WipefsArgv("/dev/x"), "ZdbLabelArgv": ZdbLabelArgv("/dev/x")}
+		"WipefsArgv": WipefsArgv("/dev/x"), "ZdbLabelArgv": ZdbLabelArgv("/dev/x"),
+		"ZdbConfigArgv": ZdbConfigArgv("tank"), "ZgenhostidArgv": ZgenhostidArgv("00bab10c")}
 	for fn, argv := range verbs {
 		for _, a := range argv {
-			if fn != "FollowArgv" && (a == "-f" || a == "-r" || a == "-R" || a == "-d") {
+			if fn != "FollowArgv" && fn != "ZgenhostidArgv" && (a == "-f" || a == "-r" || a == "-R" || a == "-d") {
 				t.Errorf("%s runs %q", fn, a)
 			}
 		}
@@ -253,7 +255,7 @@ func TestTheBundleRunsOnlyReadingVerbsSnapshotAndScrub(t *testing.T) {
 		}
 	}
 	call := regexp.MustCompile(`(\.(ask|AsRoot|Escalated|Run|Start|Output|CombinedOutput)\(|exec\.Command(Context)?\(|os\.StartProcess\(|syscall\.(Exec|ForkExec)\()`)
-	site := regexp.MustCompile(`(?:\.(?:ask|AsRoot|Escalated)|exec\.Command)\((?:\w+, )?"(zfs|zpool|systemctl|wipefs|zdb)", (\w+)\(`)
+	site := regexp.MustCompile(`(?:\.(?:ask|AsRoot|Escalated)|exec\.Command)\((?:\w+, )?"(zfs|zpool|systemctl|wipefs|zdb|zgenhostid)", (\w+)\(`)
 	// The bodies that run what they are given, each once.
 	bodies := map[string]bool{
 		"machine.go:	r := m.Run(ctx, name, args...)":              true,