Z13 zfs-hand-acts: hand acts write their key and start the run; zfs_snapshot keeps a person's snapshots apart

Builds: design 42 zfs-owns-zfs §23, §24, A-12, A-17, A-18

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

@@ -3,6 +3,9 @@ package main
 import (
 	"encoding/json"
 	"os"
+	"strings"
+	"sync"
+	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/process"
@@ -44,5 +47,75 @@ func (m *Machine) JoinBus() func() {
 	}
 	m.Emit = func(kind string, body any) error { return c.Emit(kind, body, stdio.EmitOptions{}) }
 	m.States = busStates{c}
+	m.Call = c.Call
 	return c.Close
 }
+
+// toolStates is the module's states as the tools bundle reaches them, through the tool runner.
+type toolStates struct{}
+
+func (toolStates) Put(state, key string, v any) error {
+	_, err := stdio.State(state).Put(key, v)
+	return err
+}
+func (toolStates) Delete(state, key string) error      { return stdio.State(state).Delete(key) }
+func (toolStates) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
+func (toolStates) Get(state, key string) (json.RawMessage, bool, error) {
+	e, err := stdio.State(state).Get(key)
+	if err != nil || e == nil {
+		return nil, false, err
+	}
+	return e.Value, true, nil
+}
+
+// eventHub hands the module's events, heard once through the tool runner, to the hand acts waiting for one.
+type eventHub struct {
+	mu      sync.Mutex
+	started bool
+	waiters map[int]hubWaiter
+	next    int
+}
+
+type hubWaiter struct {
+	kinds []string
+	ch    chan map[string]any
+}
+
+func (h *eventHub) await(kinds []string, bound time.Duration) (map[string]any, bool) {
+	h.mu.Lock()
+	if h.waiters == nil {
+		h.waiters = map[int]hubWaiter{}
+	}
+	h.next++
+	id, w := h.next, hubWaiter{kinds, make(chan map[string]any, 1)}
+	h.waiters[id] = w
+	first := !h.started
+	h.started = true
+	h.mu.Unlock()
+	defer func() { h.mu.Lock(); delete(h.waiters, id); h.mu.Unlock() }()
+	if first {
+		_ = stdio.Subscribe("#", func(e stdio.Envelope) error {
+			var body map[string]any
+			_ = json.Unmarshal(e.Body, &body)
+			h.mu.Lock()
+			defer h.mu.Unlock()
+			for _, w := range h.waiters {
+				for _, k := range w.kinds {
+					if e.Key == k || strings.HasSuffix(e.Key, "."+k) {
+						select {
+						case w.ch <- body:
+						default:
+						}
+					}
+				}
+			}
+			return nil
+		})
+	}
+	select {
+	case ev := <-w.ch:
+		return ev, true
+	case <-time.After(bound):
+		return nil, false
+	}
+}

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

@@ -0,0 +1,114 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+)
+
+// The hand acts (design 42 zfs-owns-zfs §23, §24; A-12, A-17, A-18).
+//
+// A hand act writes its act into the state `schedule` under its own key {by, why, at} and starts the unit of
+// the scheduled run through `<node>/node-service-manager.start` with {scope, unit}; the run reads the key,
+// acts under the same lock and guards, and deletes it. The tool waits at most HandWait for the act's event; past
+// it, it answers "started", naming the act and that `zfs_module_events` reads its outcome, and never "done" for
+// what it did not see end. The `why` is kept in the key.
+
+// HandWait is how long a hand act's tool waits for the act's event (A-12).
+const HandWait = 30 * time.Second
+
+// The acts, as the keys of the state `schedule`.
+const (
+	ActSnapshots = "snapshots"
+	ActPrune     = "prune"
+	ActApply     = "apply"
+)
+
+// handEnds are the events that end each act.
+var handEnds = map[string][]string{
+	ActSnapshots: {EventTakenPerson, EventSkippedPerson, EventRunSkipped},
+	ActPrune:     {EventRemovedPerson, EventSkippedPerson, EventRunSkipped},
+	ActApply:     {EventApplied, EventRefused, EventWaiting},
+}
+
+// StartArgs are what `node-service-manager.start` is called with to run the scheduled run now.
+func StartArgs() map[string]any { return map[string]any{"scope": "system", "unit": RunService} }
+
+// HandAnswer is what a hand act's tool answers.
+type HandAnswer struct {
+	Act string `json:"act"`
+	// Outcome is "seen" (the act's event came) or "started" (it was started, its end not seen).
+	Outcome string         `json:"outcome"`
+	Event   map[string]any `json:"event,omitempty"`
+	Said    string         `json:"said"`
+}
+
+func whyOf(why string) error {
+	switch {
+	case why == "":
+		return fmt.Errorf("say why: a hand act is a person's, and the why is kept with it")
+	case strings.ContainsAny(why, "\n\r") || len(why) > MaxWhy:
+		return fmt.Errorf("say why in one line of at most %d characters", MaxWhy)
+	}
+	return nil
+}
+
+// HandAct starts one act of the scheduled run by hand.
+func (m *Machine) HandAct(ctx context.Context, act, by, why string) (HandAnswer, error) {
+	a := HandAnswer{Act: act}
+	if _, ok := handEnds[act]; !ok {
+		return a, fmt.Errorf("%q is not an act of the run: snapshots, prune or apply", act)
+	}
+	if err := whyOf(why); err != nil {
+		return a, err
+	}
+	if m.States == nil || m.Call == nil {
+		return a, fmt.Errorf("this process reaches neither the module's state nor the service manager, so %s was not started", act)
+	}
+	if err := m.States.Put(StateSchedule, act, map[string]any{"by": by, "why": why, "at": m.Now().UTC()}); err != nil {
+		return a, fmt.Errorf("the act %s could not be written to the state %s: %v", act, StateSchedule, err)
+	}
+	if _, err := m.Call("seat:node-service-manager.start@"+m.Node, StartArgs()); err != nil {
+		_ = m.States.Delete(StateSchedule, act)
+		return a, fmt.Errorf("the service manager did not start %s: %v", RunService, err)
+	}
+	if m.Await != nil {
+		if ev, ok := m.Await(handEnds[act], HandWait); ok {
+			a.Outcome, a.Event = "seen", ev
+			a.Said = fmt.Sprintf("The act %s ran: %v.", act, ev)
+			return a, nil
+		}
+	}
+	a.Outcome = "started"
+	a.Said = fmt.Sprintf("Started the act %s (%s) and did not see it end within %s: zfs_module_events reads its outcome.", act, RunService, HandWait)
+	return a, nil
+}
+
+// Pause writes `paused` {by, why, at} in the state `schedule`: the run then takes and removes nothing (§24).
+func (m *Machine) Pause(by, why string) error {
+	if err := whyOf(why); err != nil {
+		return err
+	}
+	if m.States == nil {
+		return fmt.Errorf("this process reaches no state, so the schedule was not paused")
+	}
+	return m.States.Put(StateSchedule, KeyPaused, map[string]any{"by": by, "why": why, "at": m.Now().UTC()})
+}
+
+// Resume deletes `paused`.
+func (m *Machine) Resume() error {
+	if m.States == nil {
+		return fmt.Errorf("this process reaches no state, so the schedule was not resumed")
+	}
+	return m.States.Delete(StateSchedule, KeyPaused)
+}
+
+// Paused says whether the schedule is paused.
+func (m *Machine) Paused() bool {
+	if m.States == nil {
+		return false
+	}
+	_, ok, err := m.States.Get(StateSchedule, KeyPaused)
+	return err == nil && ok
+}

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

@@ -0,0 +1,80 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+)
+
+// §23, §24, A-12, A-17, A-18: a hand act writes its key and starts the run's unit with {scope, unit}; the tool
+// answers from the act's event when it comes within the wait, else "started", never "done"; a failed start
+// leaves no key; pause writes `paused`, resume deletes it.
+func TestAHandActWritesItsKeyStartsTheRunAndAnswersFromItsEvent(t *testing.T) {
+	cases := []struct {
+		name, act, why string
+		event          map[string]any
+		callErr        error
+		outcome        string
+		err            string
+		keyLeft        bool
+		waited         []string
+	}{
+		{name: "event seen", act: ActSnapshots, why: "before the move", event: map[string]any{"snapshot": "tank@mesh-hourly"}, outcome: "seen", keyLeft: true,
+			waited: handEnds[ActSnapshots]},
+		{name: "event not seen in time", act: ActPrune, why: "free space", outcome: "started", keyLeft: true, waited: handEnds[ActPrune]},
+		{name: "apply", act: ActApply, why: "now", event: map[string]any{"key": "hostid"}, outcome: "seen", keyLeft: true, waited: handEnds[ActApply]},
+		{name: "start refused", act: ActSnapshots, why: "x", callErr: errors.New("refused"), err: "did not start"},
+		{name: "no why", act: ActSnapshots, why: "", err: "say why"},
+		{name: "unknown act", act: "destroy", why: "x", err: "not an act"},
+	}
+	for _, c := range cases {
+		f := newFake(t)
+		m := f.machine()
+		st := memStates{}
+		m.States, m.Node = st, "ace"
+		var key string
+		var body any
+		m.Call = func(k string, b any) (json.RawMessage, error) { key, body = k, b; return nil, c.callErr }
+		var bound time.Duration
+		var kinds []string
+		m.Await = func(k []string, d time.Duration) (map[string]any, bool) {
+			kinds, bound = k, d
+			return c.event, c.event != nil
+		}
+		a, err := m.HandAct(context.Background(), c.act, "caller@node", c.why)
+		if c.err != "" {
+			if err == nil || !strings.Contains(err.Error(), c.err) {
+				t.Errorf("%s: %v", c.name, err)
+			}
+		} else if err != nil || a.Outcome != c.outcome {
+			t.Errorf("%s: %+v %v", c.name, a, err)
+		}
+		if _, left := st["schedule/"+c.act]; left != c.keyLeft {
+			t.Errorf("%s: key left %v, want %v", c.name, left, c.keyLeft)
+		}
+		if c.err == "" || c.callErr != nil {
+			args, _ := body.(map[string]any)
+			if key != "seat:node-service-manager.start@ace" || args["unit"] != "zfs-mesh-run.service" || args["scope"] != "system" || len(args) != 2 {
+				t.Errorf("%s: called %s with %v", c.name, key, body)
+			}
+		}
+		if c.err == "" && (bound != 30*time.Second || strings.Join(kinds, ",") != strings.Join(c.waited, ",")) {
+			t.Errorf("%s: waited %v for %v", c.name, bound, kinds)
+		}
+		if c.outcome == "started" && strings.Contains(a.Said, "done") {
+			t.Errorf("%s: said done for what it did not see: %s", c.name, a.Said)
+		}
+	}
+	f := newFake(t)
+	m := f.machine()
+	m.States = memStates{}
+	if err := m.Pause("p@n", "maintenance"); err != nil || !m.Paused() {
+		t.Fatal(err, m.Paused())
+	}
+	if err := m.Resume(); err != nil || m.Paused() {
+		t.Fatal(err, m.Paused())
+	}
+}

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

@@ -117,6 +117,14 @@ type Machine struct {
 	// 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)
+	// Call invokes a tool or a seat verb as this module (`seat:<seat>.<verb>@<node>`, `<module>.<tool>`); nil where
+	// this process reaches none.
+	Call func(key string, body any) (json.RawMessage, error)
+	// Node is this machine's name, for a seat verb of its own.
+	Node string
+	// Await waits at most the bound for an event of the module whose kind is one of kinds, and answers it; false
+	// when none came.
+	Await func(kinds []string, bound time.Duration) (map[string]any, bool)
 	// Warnings are what the apply saw that is no failure (a change the bus did not take, a parameter awaiting
 	// a load).
 	Warnings []string
@@ -133,7 +141,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, WriteFile: writeLive, Lock: FileLock, Errors: LoadErrorMemory(ErrorsFile)}
+	return &Machine{Run: ExecRunner, UID: os.Getuid(), Now: time.Now, ReadFile: os.ReadFile, Resolve: filepath.EvalSymlinks, Node: os.Getenv("MESH_NODE"), 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

@@ -67,6 +67,7 @@ func main() {
 	}
 	m := Here()
 	m.Emit = func(eventType string, body any) error { return stdio.Emit(eventType, body) }
+	m.States, m.Call, m.Await = toolStates{}, stdio.Ask, (&eventHub{}).await
 	// An empty name serves as the module the runtime names (MESH_SERVED_MODULE): zfs.
 	if err := stdio.Serve("", Tools(m)); err != nil {
 		fmt.Fprintln(os.Stderr, err)

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

@@ -35,19 +35,23 @@ func readManifest(t *testing.T) (manifest, string) {
 // The manifest lists exactly the tools the bundle serves; every one but zfs_snapshot says it only reads,
 // and zfs_snapshot says it acts. Each says what it replaces (novox/hq ADR 0245), but the health check,
 // which replaces nothing. No seat: a new mesh or node seat is the controller's to define (ADR 0121).
+// handTools are the tools that start or steer the scheduled run, which replace no command.
+var handTools = map[string]bool{"zfs_snapshots_run_now": true, "zfs_snapshots_prune_now": true, "zfs_config_apply": true,
+	"zfs_snapshots_pause": true, "zfs_snapshots_resume": true}
+
 func TestTheManifestNamesTheToolsTheBundleServesAndWhatEachReplaces(t *testing.T) {
 	m, raw := readManifest(t)
 	var served []string
 	for _, tl := range Tools(newFake(t).machine()) {
 		served = append(served, tl.Name)
 		want := "(r)"
-		if tl.Name == "zfs_snapshot" || strings.HasPrefix(tl.Name, "zfs_scrub_") {
+		if tl.Name == "zfs_snapshot" || strings.HasPrefix(tl.Name, "zfs_scrub_") || handTools[tl.Name] {
 			want = "(a)"
 		}
 		if !strings.HasSuffix(tl.Description, want) {
 			t.Errorf("%s does not end with %s", tl.Name, want)
 		}
-		if tl.Name != "zfs_health" && len(m.Replaces[tl.Name]) == 0 {
+		if tl.Name != "zfs_health" && !handTools[tl.Name] && len(m.Replaces[tl.Name]) == 0 {
 			t.Errorf("%s says nothing of what it replaces", tl.Name)
 		}
 		for _, c := range m.Replaces[tl.Name] {
@@ -68,7 +72,7 @@ func TestTheManifestNamesTheToolsTheBundleServesAndWhatEachReplaces(t *testing.T
 	if len(m.Claims) != 0 {
 		t.Errorf("the module claims no seat: %v", m.Claims)
 	}
-	if strings.Join(m.Emits, ",") != SnapshotEvent+","+ScrubEvent {
+	if strings.Join(m.Emits, ",") != "snapshot.taken,"+ScrubEvent {
 		t.Errorf("emits %v", m.Emits)
 	}
 	if hit := regexp.MustCompile(`\b(g14|novox|ace|shanks|jochen)\b|/home/`).FindString(strings.ReplaceAll(raw, "novox/hq", "")); hit != "" {

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

@@ -89,14 +89,14 @@ func TestTheSnapshotHasNoRecursiveAndSaysWhatItWillHold(t *testing.T) {
 func TestASnapshotThatTimedOutIsOutcomeUnknownUnlessTheListShowsIt(t *testing.T) {
 	f := withZFS(t)
 	f.answers[takeLine] = Ran{Code: -1, TimedOut: true, Err: "did not answer within its time"}
-	_, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why")
+	_, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why", "tester")
 	if err == nil || !strings.Contains(err.Error(), "outcome unknown; check zfs_snapshots") || strings.Contains(err.Error(), "NOT DONE") {
 		t.Errorf("timed out, not shown: %v", err)
 	}
 	f = withZFS(t)
 	snapshotting(f, takeLine, "tank/media@before-move\t1791400000\t0\t1")
 	f.answers[takeLine] = Ran{Code: -1, TimedOut: true}
-	a, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why")
+	a, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why", "tester")
 	if err != nil || !a.Done {
 		t.Errorf("timed out and shown by the list: done (%v, %+v)", err, a)
 	}

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

@@ -9,9 +9,9 @@ import (
 	"time"
 )
 
-// SnapshotEvent is the event each attempted snapshot is published as (module.json's emits): the record
-// of the act, beside the pool's own history, which ZFS writes for every snapshot.
-const SnapshotEvent = "snapshot.taken"
+// MeshPrefix begins the name of every ZFS snapshot the scheduled run takes (§21): a person's snapshot never
+// does, so the run's snapshots and the person's stay apart (§23).
+const MeshPrefix = "mesh-"
 
 // The one act of this module: a snapshot of one dataset, by a person, with why. It adds a snapshot and
 // changes nothing else: no destroy, rollback, set or send exists in this module (novox/hq ADR 0051, the
@@ -71,7 +71,7 @@ type SnapshotAnswer struct {
 // TakeSnapshot takes dataset@name. It refuses an unknown dataset, a name with anything but
 // [A-Za-z0-9._-], an empty why, and a snapshot that already exists. It answers what `zfs list` shows
 // afterwards.
-func (m *Machine) TakeSnapshot(ctx context.Context, dataset, name, why string) (SnapshotAnswer, error) {
+func (m *Machine) TakeSnapshot(ctx context.Context, dataset, name, why, by string) (SnapshotAnswer, error) {
 	a := SnapshotAnswer{Dataset: dataset, Name: name, Why: why, Made: []Made{}, Others: []string{}, Pool: poolOf(dataset)}
 	switch {
 	case dataset == "":
@@ -80,6 +80,8 @@ func (m *Machine) TakeSnapshot(ctx context.Context, dataset, name, why string) (
 		return a, fmt.Errorf("%q is not a dataset: name the filesystem or volume, and the snapshot's name in name", dataset)
 	case name == "":
 		return a, fmt.Errorf("name the snapshot")
+	case strings.HasPrefix(name, MeshPrefix):
+		return a, fmt.Errorf("the snapshot's name may not begin %q: those are the scheduled run's, and a person's snapshot stays apart from them", MeshPrefix)
 	case !snapName.MatchString(name):
 		return a, fmt.Errorf("the snapshot's name %q may hold only letters, digits, '.', '_' and '-'", name)
 	case len(dataset)+1+len(name) > MaxName:
@@ -176,7 +178,7 @@ func (m *Machine) TakeSnapshot(ctx context.Context, dataset, name, why string) (
 		a.Said += " " + space + " " + a.Holds + "."
 	}
 
-	a.Recorded, a.Record = m.record(a)
+	a.Recorded, a.Record = m.record(a, by)
 	if !a.Recorded {
 		a.Said += " NOT RECORDED on the bus: " + a.Record + "; the pool's own history (zpool history) has the command."
 	}
@@ -188,7 +190,7 @@ func (m *Machine) TakeSnapshot(ctx context.Context, dataset, name, why string) (
 }
 
 // record publishes the act as this module's event: the names, the reason and the outcome, nothing else.
-func (m *Machine) record(a SnapshotAnswer) (bool, string) {
+func (m *Machine) record(a SnapshotAnswer, by string) (bool, string) {
 	if m.Emit == nil {
 		return false, "this process publishes no events"
 	}
@@ -196,13 +198,13 @@ func (m *Machine) record(a SnapshotAnswer) (bool, string) {
 	if len(a.Made) > 0 {
 		created = a.Made[0].Created
 	}
-	err := m.Emit(SnapshotEvent, map[string]any{"dataset": a.Dataset, "snapshot": a.Name,
+	err := m.Emit(EventTakenPerson, map[string]any{"dataset": a.Dataset, "snapshot": a.Dataset + "@" + a.Name, "by": by,
 		"why": a.Why, "done": a.Done, "outcome": a.Outcome, "made": madeNames(a.Made), "created": created,
 		"at": m.Now().UTC().Format(time.RFC3339)})
 	if err != nil {
 		return false, err.Error()
 	}
-	return true, "published as the event " + SnapshotEvent
+	return true, "published as the event " + EventTakenPerson
 }
 
 // spaceAfter reads the dataset's free space and its pool's capacity after the act, and says them.

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

@@ -39,7 +39,7 @@ func TestTheSnapshotCommandIsFixedWordsThroughSudoWithTheNameAfterDashDash(t *te
 func TestASnapshotIsAnsweredFromTheListAfterwardsAndRecorded(t *testing.T) {
 	f := withZFS(t)
 	snapshotting(f, takeLine, "tank/media@before-move\t1791400000\t0\t26938034880512")
-	a, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "before moving the library")
+	a, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "before moving the library", "tester")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -55,7 +55,7 @@ func TestASnapshotIsAnsweredFromTheListAfterwardsAndRecorded(t *testing.T) {
 	if strings.Join(a.Ran, " ") != takeLine || len(ran(f, "sudo")) != 1 {
 		t.Errorf("ran: %v; every call: %v", a.Ran, f.calls)
 	}
-	if len(f.emitted) != 1 || f.emitted[0]["_type"] != SnapshotEvent || f.emitted[0]["why"] != "before moving the library" ||
+	if len(f.emitted) != 1 || f.emitted[0]["_type"] != EventTakenPerson || f.emitted[0]["why"] != "before moving the library" ||
 		f.emitted[0]["done"] != true || f.emitted[0]["made"] != "tank/media@before-move" || !a.Recorded {
 		t.Errorf("recorded: %+v", f.emitted)
 	}
@@ -64,7 +64,7 @@ func TestASnapshotIsAnsweredFromTheListAfterwardsAndRecorded(t *testing.T) {
 func TestASnapshotTheListDoesNotShowIsNotAnsweredAsMade(t *testing.T) {
 	f := withZFS(t)
 	f.on(takeLine, "") // zfs exits 0, and the list does not change
-	a, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why")
+	a, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why", "tester")
 	if err == nil || a.Done || !strings.Contains(err.Error(), "the list shows 0 of the 1") {
 		t.Fatalf("err %v, answer %+v", err, a)
 	}
@@ -76,7 +76,7 @@ func TestASnapshotTheListDoesNotShowIsNotAnsweredAsMade(t *testing.T) {
 func TestARefusedSudoIsSaidAndRecorded(t *testing.T) {
 	f := withZFS(t)
 	f.answers[takeLine] = Ran{Code: 1, Err: "sudo: a password is required"}
-	_, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why")
+	_, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why", "tester")
 	if err == nil || !strings.Contains(err.Error(), "NOT DONE") || !strings.Contains(err.Error(), "sudo refused the operator account") {
 		t.Fatalf("err: %v", err)
 	}
@@ -89,7 +89,7 @@ func TestAnUnrecordedActSaysSoLoudly(t *testing.T) {
 	f := withZFS(t)
 	snapshotting(f, takeLine, "tank/media@before-move\t1791400000\t0\t1")
 	f.emitErr = errors.New("the bus did not accept it")
-	a, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why")
+	a, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why", "tester")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -101,7 +101,7 @@ func TestAnUnrecordedActSaysSoLoudly(t *testing.T) {
 func TestAnotherSnapshotMeanwhileIsSaidNotHidden(t *testing.T) {
 	f := withZFS(t)
 	snapshotting(f, takeLine, "tank/media@before-move\t1791400000\t0\t1", "tank/downloads@auto-1200\t1791547201\t0\t1")
-	a, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why")
+	a, err := f.machine().TakeSnapshot(context.Background(), "tank/media", "before-move", "why", "tester")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -116,6 +116,7 @@ func TestEveryRefusalRunsNoSnapshotCommand(t *testing.T) {
 		{"", "x", "why", "name the dataset"},
 		{"tank/media@x", "y", "why", "is not a dataset"},
 		{"tank/media", "", "why", "name the snapshot"},
+		{"tank/media", "mesh-hourly-20261109T1200Z", "why", "may not begin"},
 		{"tank/media", "a b", "why", "may hold only"},
 		{"tank/media", "x;rm -rf /", "why", "may hold only"},
 		{"tank/media", "$(reboot)", "why", "may hold only"},
@@ -129,7 +130,7 @@ func TestEveryRefusalRunsNoSnapshotCommand(t *testing.T) {
 	}
 	for _, c := range cases {
 		f := withZFS(t)
-		_, err := f.machine().TakeSnapshot(context.Background(), c.dataset, c.name, c.why)
+		_, err := f.machine().TakeSnapshot(context.Background(), c.dataset, c.name, c.why, "tester")
 		if c.says == "" {
 			if len(ran(f, "sudo -n zfs snapshot -- tank/media@-r")) != 1 {
 				t.Errorf("%q: the name follows -- and runs as a name: %v", c.name, f.calls)
@@ -149,7 +150,7 @@ func TestAMachineWithoutZFSSaysTheModuleDoesNotApply(t *testing.T) {
 	f := newFake(t) // every command: executable file not found
 	m := f.machine()
 	for _, tool := range Tools(m) {
-		if tool.Name == "zfs_health" {
+		if tool.Name == "zfs_health" || handTools[tool.Name] {
 			continue
 		}
 		args := map[string]any{"dataset": "tank", "name": "x", "why": "y"}

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

@@ -41,7 +41,7 @@ func Tools(m *Machine) []stdio.Tool {
 				"answers what zfs list shows: the snapshot made, its creation time, any other snapshot that came or went " +
 				"meanwhile, the dataset's free space and the pool's capacity. From then on the snapshot holds every block " +
 				"of the dataset deleted or overwritten, until it is destroyed by hand. A command that ran out of time is " +
-				"answered from the list, or as outcome unknown. Recorded as the event snapshot.taken; ZFS keeps it in the " +
+				"answered from the list, or as outcome unknown. A name beginning mesh- is refused (the scheduled run's). Recorded as the event snapshot.taken.person; ZFS keeps it in the " +
 				"pool's history too. It adds a snapshot and changes nothing else. Replaces zfs snapshot. (a)",
 			Input: map[string]any{
 				"dataset": dataset("the dataset, as zfs_datasets names it"),
@@ -52,7 +52,7 @@ func Tools(m *Machine) []stdio.Tool {
 				if _, given := a["recursive"]; given {
 					return nil, fmt.Errorf("recursive does not exist in this version: zfs_snapshot snapshots one dataset, never its children")
 				}
-				ans, err := m.TakeSnapshot(ctx, text(a, "dataset"), text(a, "name"), text(a, "why"))
+				ans, err := m.TakeSnapshot(ctx, text(a, "dataset"), text(a, "name"), text(a, "why"), stdio.CallerOf(a))
 				if err != nil {
 					if _, absentHere := err.(errAbsent); absentHere {
 						return NotApplicable{Applicable: false, Said: err.Error()}, nil
@@ -62,6 +62,30 @@ func Tools(m *Machine) []stdio.Tool {
 				return ans, nil
 			},
 		},
+		handTool(m, "zfs_snapshots_run_now", ActSnapshots, "Take the scheduled ZFS snapshots now, as the scheduled run does (a person's act with why): writes the act into the state `schedule` and starts zfs-mesh-run.service through the service manager; the run takes it under the same lock and guards. Waits up to 30 seconds for the act's event, else answers started and says zfs_module_events reads its outcome. (a)"),
+		handTool(m, "zfs_snapshots_prune_now", ActPrune, "Remove the scheduled ZFS snapshots past their keep count now, as the scheduled run does, only those its own ledger holds (a person's act with why). Waits up to 30 seconds for the act's event, else answers started and says zfs_module_events reads its outcome. (a)"),
+		handTool(m, "zfs_config_apply", ActApply, "Converge the node to its config file now, as the apply does (a person's act with why): the same lock and guards as the apply. Waits up to 30 seconds for the act's event, else answers started and says zfs_module_events reads its outcome. (a)"),
+		{
+			Name:        "zfs_snapshots_pause",
+			Description: "Pause the scheduled ZFS snapshots: writes `paused` {by, why, at} in the state `schedule`; the run then takes and removes nothing until zfs_snapshots_resume. (a)",
+			Input:       map[string]any{"why": map[string]any{"type": "string", "description": "why, in one line; kept with the act"}},
+			Run: func(a map[string]any) (any, error) {
+				if err := m.Pause(stdio.CallerOf(a), text(a, "why")); err != nil {
+					return nil, err
+				}
+				return map[string]any{"paused": true}, nil
+			},
+		},
+		{
+			Name:        "zfs_snapshots_resume",
+			Description: "Resume the scheduled ZFS snapshots: deletes `paused` from the state `schedule`. (a)",
+			Run: func(map[string]any) (any, error) {
+				if err := m.Resume(); err != nil {
+					return nil, err
+				}
+				return map[string]any{"paused": false}, nil
+			},
+		},
 		{
 			Name: "zfs_pools",
 			Description: "Every imported pool's health in a line: its state (ONLINE, DEGRADED, FAULTED, …) with status's own " +
@@ -255,3 +279,15 @@ func whole(a map[string]any, key string, fallback, least, most int) (int, error)
 	}
 	return n, nil
 }
+
+// handTool is one hand act as a tool.
+func handTool(m *Machine, name, act, description string) stdio.Tool {
+	return stdio.Tool{
+		Name:        name,
+		Description: description,
+		Input:       map[string]any{"why": map[string]any{"type": "string", "description": "why, in one line; kept with the act"}},
+		Run: func(a map[string]any) (any, error) {
+			return m.HandAct(context.Background(), act, stdio.CallerOf(a), text(a, "why"))
+		},
+	}
+}

modules/zfs/module.json

@@ -59,6 +59,11 @@
     "zfs_scrub_start",
     "zfs_scrub_pause",
     "zfs_scrub_stop",
+    "zfs_snapshots_run_now",
+    "zfs_snapshots_prune_now",
+    "zfs_config_apply",
+    "zfs_snapshots_pause",
+    "zfs_snapshots_resume",
     "zfs_io",
     "zfs_arc",
     "zfs_events",