diff --git a/modules/zfs/cmd/zfs-tools/apply_properties.go b/modules/zfs/cmd/zfs-tools/apply_properties.go
new file mode 100644
index 000000000..495bf4ffa
--- /dev/null
+++ b/modules/zfs/cmd/zfs-tools/apply_properties.go
@@ -0,0 +1,176 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+// The apply's pool and dataset properties (design 42 zfs-owns-zfs §16, §10).
+//
+// For each set pool property the apply runs `zpool set <property>=<value> <pool>`, and for `default` `zpool
+// set <property>=<ZFS's default> <pool>`. For each set dataset property it runs `zfs set <property>=<value>
+// <dataset>`, and for `default` `zfs inherit <property> <dataset>`, so its source stays default. A property
+// already at its value runs nothing. Each change is read back with `zpool get -Hp` or `zfs get -Hp` and said.
+// sharenfs and sharesmb are never here: the parse refuses them (config.go).
+
+// poolDefaults are ZFS's defaults of the pool properties the module sets.
+var poolDefaults = map[string]string{"autotrim": "off", "autoexpand": "off", "autoreplace": "off", "failmode": "wait"}
+
+// PoolGetArgv reads one pool property's value.
+func PoolGetArgv(pool, property string) []string {
+	return []string{"get", "-Hp", "-o", "value", property, "--", pool}
+}
+
+// PoolSetArgv sets one pool property.
+func PoolSetArgv(pool, property, value string) []string {
+	return []string{"set", property + "=" + value, "--", pool}
+}
+
+// DatasetGetArgv reads one dataset property's value and source.
+func DatasetGetArgv(dataset, property string) []string {
+	return []string{"get", "-Hp", "-o", "value,source", property, "--", dataset}
+}
+
+// DatasetSetArgv sets one dataset property.
+func DatasetSetArgv(dataset, property, value string) []string {
+	return []string{"set", property + "=" + value, "--", dataset}
+}
+
+// DatasetInheritArgv gives a dataset property back to ZFS's inheritance.
+func DatasetInheritArgv(dataset, property string) []string {
+	return []string{"inherit", property, "--", dataset}
+}
+
+// sizeBytes reads `1M`, `128K`, `5T` or plain digits as bytes; false for another form.
+func sizeBytes(v string) (uint64, bool) {
+	mult := uint64(1)
+	if n := len(v); n > 0 {
+		if i := strings.IndexByte("KMGT", v[n-1]); i >= 0 {
+			mult = uint64(1) << (10 * (i + 1))
+			v = v[:n-1]
+		}
+	}
+	n, err := strconv.ParseUint(v, 10, 64)
+	return n * mult, err == nil
+}
+
+// sameValue says whether ZFS's `-p` value is the wanted one in the form the config states it.
+func sameValue(property, got, want string) bool {
+	switch property {
+	case "recordsize", "quota", "reservation":
+		if want == "none" {
+			return got == "0" || got == "none"
+		}
+		g, ok1 := sizeBytes(got)
+		w, ok2 := sizeBytes(want)
+		return ok1 && ok2 && g == w
+	}
+	return got == want
+}
+
+// ApplyProperties converges the set pool and dataset properties. It answers every command that failed, as
+// 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)
+	}
+	for _, p := range sortedKeys(c.PoolProperties) {
+		key := "pool-properties." + p
+		want := c.PoolProperties[p]
+		if want == "default" {
+			want = poolDefaults[p]
+		}
+		got, err := m.ask(ctx, "zpool", PoolGetArgv(c.Pool, p)...)
+		if err != nil {
+			fail(key, err)
+			continue
+		}
+		before := strings.TrimSpace(got)
+		if before == want {
+			_ = m.keepAsSet(key, before)
+			continue
+		}
+		if _, err := m.AsRoot(ctx, "zpool", PoolSetArgv(c.Pool, p, want)...); err != nil {
+			fail(key, err)
+			continue
+		}
+		back, err := m.ask(ctx, "zpool", PoolGetArgv(c.Pool, p)...)
+		if err != nil || strings.TrimSpace(back) != want {
+			fail(key, fmt.Errorf("read back %q after setting %q (%v)", strings.TrimSpace(back), want, err))
+			continue
+		}
+		if err := m.Changed("pool", key, before, want, strings.TrimSpace(back)); err != nil {
+			fail(key, err)
+		}
+	}
+	for _, ds := range sortedKeys(c.DatasetProperties) {
+		for _, p := range sortedKeys(c.DatasetProperties[ds]) {
+			key := "dataset-properties." + ds + "." + p
+			want := c.DatasetProperties[ds][p]
+			value, source, err := m.datasetProperty(ctx, ds, p)
+			if err != nil {
+				fail(key, err)
+				continue
+			}
+			before := value
+			if want == "default" {
+				if source != "local" && source != "received" {
+					_ = m.keepAsSet(key, before)
+					continue
+				}
+				if _, err := m.AsRoot(ctx, "zfs", DatasetInheritArgv(ds, p)...); err != nil {
+					fail(key, err)
+					continue
+				}
+				back, bsource, err := m.datasetProperty(ctx, ds, p)
+				if err != nil || bsource == "local" || bsource == "received" {
+					fail(key, fmt.Errorf("read back source %q after inheriting (%v)", bsource, err))
+					continue
+				}
+				if err := m.Changed("dataset", key, before, "default", back); err != nil {
+					fail(key, err)
+				}
+				continue
+			}
+			if sameValue(p, value, want) {
+				_ = m.keepAsSet(key, before)
+				continue
+			}
+			if _, err := m.AsRoot(ctx, "zfs", DatasetSetArgv(ds, p, want)...); err != nil {
+				fail(key, err)
+				continue
+			}
+			back, _, err := m.datasetProperty(ctx, ds, p)
+			if err != nil || !sameValue(p, back, want) {
+				fail(key, fmt.Errorf("read back %q after setting %q (%v)", back, want, err))
+				continue
+			}
+			if err := m.Changed("dataset", key, before, want, back); err != nil {
+				fail(key, err)
+			}
+		}
+	}
+	return errors.Join(failed...)
+}
+
+// keepAsSet keeps a value found already at what is set: no command, no event, the state says it is as set.
+func (m *Machine) keepAsSet(key, value string) error {
+	return m.keep(key, AppliedEntry{Before: value, After: value, Outcome: OutcomeApplied})
+}
+
+func (m *Machine) datasetProperty(ctx context.Context, ds, p string) (value, source string, err error) {
+	out, err := m.ask(ctx, "zfs", DatasetGetArgv(ds, p)...)
+	if err != nil {
+		return "", "", err
+	}
+	f := strings.Split(strings.TrimSpace(out), "\t")
+	if len(f) != 2 {
+		return "", "", fmt.Errorf("zfs get answered %q, not a value and a source", strings.TrimSpace(out))
+	}
+	return f[0], f[1], nil
+}
diff --git a/modules/zfs/cmd/zfs-tools/apply_properties_test.go b/modules/zfs/cmd/zfs-tools/apply_properties_test.go
new file mode 100644
index 000000000..12b897e2c
--- /dev/null
+++ b/modules/zfs/cmd/zfs-tools/apply_properties_test.go
@@ -0,0 +1,100 @@
+package main
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+// §16, §31 convergence table: a property at its value runs nothing; a set one is set and read back and
+// said; `default` is ZFS's default for a pool property and an inherit for a dataset's; a read-back that
+// differs is a failure.
+func TestPropertiesConvergeFromWhatZFSSays(t *testing.T) {
+	cfg := func(j string) *Config {
+		c, ref := ParseConfig([]byte(j), 16000)
+		if len(ref) != 0 {
+			t.Fatal(ref)
+		}
+		return c
+	}
+	type step struct{ cmd, out string }
+	cases := []struct {
+		name   string
+		conf   string
+		reads  []step // read answers
+		writes []string
+		after  map[string]string // a read that changes once the write ran
+		events int
+		fails  bool
+	}{
+		{"pool property at its value", `{"pool":"tank","pool-properties":{"autotrim":"on"}}`,
+			[]step{{"zpool get -Hp -o value autotrim -- tank", "on\n"}}, nil, nil, 0, false},
+		{"pool property set", `{"pool":"tank","pool-properties":{"autotrim":"on"}}`,
+			[]step{{"zpool get -Hp -o value autotrim -- tank", "off\n"}}, []string{"zpool set autotrim=on -- tank"},
+			map[string]string{"zpool get -Hp -o value autotrim -- tank": "on\n"}, 1, false},
+		{"pool property to ZFS's default", `{"pool":"tank","pool-properties":{"failmode":"default"}}`,
+			[]step{{"zpool get -Hp -o value failmode -- tank", "continue\n"}}, []string{"zpool set failmode=wait -- tank"},
+			map[string]string{"zpool get -Hp -o value failmode -- tank": "wait\n"}, 1, false},
+		{"pool property read back differs", `{"pool":"tank","pool-properties":{"autotrim":"on"}}`,
+			[]step{{"zpool get -Hp -o value autotrim -- tank", "off\n"}}, []string{"zpool set autotrim=on -- tank"}, nil, 0, true},
+		{"dataset size at its value in bytes", `{"pool":"tank","dataset-properties":{"tank/a":{"recordsize":"1M"}}}`,
+			[]step{{"zfs get -Hp -o value,source recordsize -- tank/a", "1048576\tlocal\n"}}, nil, nil, 0, false},
+		{"dataset compression set", `{"pool":"tank","dataset-properties":{"tank":{"compression":"zstd-3"}}}`,
+			[]step{{"zfs get -Hp -o value,source compression -- tank", "lz4\tlocal\n"}}, []string{"zfs set compression=zstd-3 -- tank"},
+			map[string]string{"zfs get -Hp -o value,source compression -- tank": "zstd-3\tlocal\n"}, 1, false},
+		{"dataset default of an inherited one", `{"pool":"tank","dataset-properties":{"tank/a":{"atime":"default"}}}`,
+			[]step{{"zfs get -Hp -o value,source atime -- tank/a", "on\tinherited from tank\n"}}, nil, nil, 0, false},
+		{"dataset default of a local one", `{"pool":"tank","dataset-properties":{"tank/a":{"atime":"default"}}}`,
+			[]step{{"zfs get -Hp -o value,source atime -- tank/a", "off\tlocal\n"}}, []string{"zfs inherit atime -- tank/a"},
+			map[string]string{"zfs get -Hp -o value,source atime -- tank/a": "on\tdefault\n"}, 1, false},
+		{"quota none at zero", `{"pool":"tank","dataset-properties":{"tank/a":{"quota":"none"}}}`,
+			[]step{{"zfs get -Hp -o value,source quota -- tank/a", "0\tdefault\n"}}, nil, nil, 0, false},
+	}
+	for _, c := range cases {
+		f := newFake(t)
+		f.uid = 0
+		for _, r := range c.reads {
+			f.on(r.cmd, r.out)
+		}
+		wrote := map[string]bool{}
+		for _, w := range c.writes {
+			w := w
+			f.on(w, "")
+			f.after[w] = func(f *fake) {
+				wrote[w] = true
+				for k, v := range c.after {
+					f.on(k, v)
+				}
+			}
+		}
+		m := f.machine()
+		m.States = memStates{}
+		err := m.ApplyProperties(context.Background(), cfg(c.conf))
+		if (err != nil) != c.fails {
+			t.Errorf("%s: error %v", c.name, err)
+		}
+		for _, w := range c.writes {
+			if !wrote[w] {
+				t.Errorf("%s: %q not run; ran %v", c.name, w, f.calls)
+			}
+		}
+		for _, call := range f.calls {
+			if (strings.Contains(call, " set ") || strings.Contains(call, " inherit ")) && !contains(c.writes, call) {
+				t.Errorf("%s: ran %q", c.name, call)
+			}
+		}
+		if len(f.emitted) != c.events {
+			t.Errorf("%s: %d events, want %d: %v", c.name, len(f.emitted), c.events, f.emitted)
+		}
+	}
+}
+
+// §10: a dataset's sharenfs and sharesmb are never converged, never run: the parse refuses them, so the
+// apply has no Config to run.
+func TestSharenfsAndSharesmbAreNeverASetProperty(t *testing.T) {
+	for _, p := range []string{"sharenfs", "sharesmb"} {
+		if c, ref := ParseConfig([]byte(`{"pool":"tank","dataset-properties":{"tank":{"`+p+`":"on"}}}`), 16000); c != nil || len(ref) == 0 {
+			t.Errorf("%s accepted", p)
+		}
+	}
+}
diff --git a/modules/zfs/cmd/zfs-tools/manifest_test.go b/modules/zfs/cmd/zfs-tools/manifest_test.go
index b917c57a1..3814ce6ae 100644
--- a/modules/zfs/cmd/zfs-tools/manifest_test.go
+++ b/modules/zfs/cmd/zfs-tools/manifest_test.go
@@ -218,12 +218,22 @@ func TestTheBundleRunsOnlyReadingVerbsSnapshotAndScrub(t *testing.T) {
 		"EventsArgv": {"zpool", "events"}, "HistoryArgv": {"zpool", "history"}, "FollowArgv": {"zpool", "events"},
 		"ScrubArgv": {"zpool", "scrub"}, "WaitArgv": {"zpool", "wait"},
 		"TimerShowArgv": {"systemctl", "show"}, "ServiceShowArgv": {"systemctl", "show"},
+		"PoolGetArgv": {"zpool", "get"}, "PoolSetArgv": {"zpool", "set"}, "DatasetGetArgv": {"zfs", "get"},
+		"DatasetSetArgv": {"zfs", "set"}, "DatasetInheritArgv": {"zfs", "inherit"},
 	}
 	verbs := map[string][]string{"DatasetsArgv": DatasetsArgv(), "SnapshotsArgv": SnapshotsArgv(), "PoolsArgv": PoolsArgv(),
 		"StatusArgv": StatusArgv(), "IOArgv": IOArgv(3), "EventsArgv": EventsArgv(), "HistoryArgv": HistoryArgv("tank"),
 		"PropertiesArgv": PropertiesArgv("tank"), "SnapshotArgv": SnapshotArgv("tank", "x"), "FollowArgv": FollowArgv(),
-		"TimerShowArgv": TimerShowArgv("x.timer"), "ServiceShowArgv": ServiceShowArgv("x.service"), "WaitArgv": WaitArgv("tank")}
+		"TimerShowArgv": TimerShowArgv("x.timer"), "ServiceShowArgv": ServiceShowArgv("x.service"), "WaitArgv": WaitArgv("tank"),
+		"PoolGetArgv": PoolGetArgv("tank", "autotrim"), "PoolSetArgv": PoolSetArgv("tank", "autotrim", "on"),
+		"DatasetGetArgv": DatasetGetArgv("tank", "atime"), "DatasetSetArgv": DatasetSetArgv("tank", "atime", "on"),
+		"DatasetInheritArgv": DatasetInheritArgv("tank", "atime")}
 	for fn, argv := range verbs {
+		for _, a := range argv {
+			if fn != "FollowArgv" && (a == "-f" || a == "-r" || a == "-R" || a == "-d") {
+				t.Errorf("%s runs %q", fn, a)
+			}
+		}
 		if argv[0] != allowed[fn][1] {
 			t.Errorf("%s runs %q, not %q", fn, argv[0], allowed[fn][1])
 		}