Z6 zfs-events-state: each change, refusal and wait said as its event and kept in the state applied

Builds: design 42 zfs-owns-zfs §25, §26, A-19
2026-11-13 00:26:32 +01:00
parent 3cb2b83a15
commit 0fa4b5d8e7
3 changed files with 199 additions and 0 deletions
@@ -0,0 +1,131 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+// What the module says and keeps of what it changed (design 42 zfs-owns-zfs §25, §26).
+//
+// Each kind of event has a subject of its own on the module's events stream (the tool runner's or the
+// process's `Emit`, by the kind's name). The state `applied` holds one key per config value, `{before, after,
+// at, outcome}`, written by the apply and deleted by it when the value goes back to not managed, after its
+// last `config.applied`.
+
+// The module's event kinds (§25), as the subject's last words.
+const (
+	EventApplied        = "config.applied"
+	EventRefused        = "config.refused"
+	EventWaiting        = "config.waiting"
+	EventTakenPerson    = "snapshot.taken.person"
+	EventTakenSchedule  = "snapshot.taken.schedule"
+	EventRemovedPerson  = "snapshot.removed.person"
+	EventRemovedSched   = "snapshot.removed.schedule"
+	EventSkippedPerson  = "snapshot.skipped.person"
+	EventSkippedSched   = "snapshot.skipped.schedule"
+	EventRunSkipped     = "run.skipped"
+	EventScrubRequested = ScrubEvent
+)
+
+// The module's states (§26): the state's name, as `requires` declares it.
+const (
+	StateApplied  = "applied"
+	StateSchedule = "schedule"
+	StateFindings = "findings"
+)
+
+// The outcomes a key of the state `applied` holds.
+const (
+	OutcomeApplied = "applied"
+	OutcomeRefused = "refused"
+	OutcomeWaiting = "waiting"
+	OutcomeFailed  = "failed"
+)
+
+// StateStore is the module's states on the bus, as the process or the tools bundle reaches them.
+type StateStore interface {
+	Put(state, key string, value any) error
+	Delete(state, key string) error
+	// Get answers false when the key is not there.
+	Get(state, key string) (json.RawMessage, bool, error)
+}
+
+// AppliedEntry is one key of the state `applied`.
+type AppliedEntry struct {
+	Before  string    `json:"before"`
+	After   string    `json:"after"`
+	At      time.Time `json:"at"`
+	Outcome string    `json:"outcome"`
+	// Why is the reason of a refusal or a failure, WaitsOn what a wait waits on, Since when it began.
+	Why     string    `json:"why,omitempty"`
+	WaitsOn string    `json:"waits-on,omitempty"`
+	Since   time.Time `json:"since,omitzero"`
+}
+
+// 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 m.Emit(kind, body)
+}
+
+// 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)
+	}
+	e.At = m.Now().UTC()
+	return m.States.Put(StateApplied, key, e)
+}
+
+// Changed says a change as `config.applied` {target, key, before, after, read-back} and keeps it. The event
+// goes first: the state is deleted only after the last event of its key.
+func (m *Machine) Changed(target, key, before, after, readBack string) error {
+	if err := m.say(EventApplied, map[string]any{"target": target, "key": key, "before": before, "after": after,
+		"read-back": readBack}); err != nil {
+		return err
+	}
+	return m.keep(key, AppliedEntry{Before: before, After: after, Outcome: OutcomeApplied})
+}
+
+// Refused says a refusal as `config.refused` {key, value, reason} and keeps it, so zfs_health is unhealthy
+// naming the key (§29).
+func (m *Machine) Refused(r Refusal) error {
+	if err := m.say(EventRefused, map[string]any{"key": r.Key, "value": r.Value, "reason": r.Reason}); err != nil {
+		return err
+	}
+	return m.keep(r.Key, AppliedEntry{After: r.Value, Outcome: OutcomeRefused, Why: r.Reason})
+}
+
+// Waiting says a wait as `config.waiting` {key, waits-on, since} and keeps it; a wait already kept keeps
+// its own since.
+func (m *Machine) Waiting(key, waitsOn string) error {
+	since := m.Now().UTC()
+	if m.States != nil {
+		if raw, ok, err := m.States.Get(StateApplied, key); err == nil && ok {
+			var old AppliedEntry
+			if json.Unmarshal(raw, &old) == nil && old.Outcome == OutcomeWaiting && old.WaitsOn == waitsOn && !old.Since.IsZero() {
+				since = old.Since
+			}
+		}
+	}
+	if err := m.say(EventWaiting, map[string]any{"key": key, "waits-on": waitsOn, "since": since.Format(time.RFC3339)}); err != nil {
+		return err
+	}
+	return m.keep(key, AppliedEntry{Outcome: OutcomeWaiting, WaitsOn: waitsOn, Since: since})
+}
+
+// CommandFailed keeps a failed command under its key; the apply then exits non-zero (A-8).
+func (m *Machine) CommandFailed(key string, cause error) error {
+	return m.keep(key, AppliedEntry{Outcome: OutcomeFailed, Why: cause.Error()})
+}
+
+// NotManaged is a value gone back to not managed: its key leaves the state, after its last `config.applied`.
+func (m *Machine) NotManaged(key string) error {
+	if m.States == nil {
+		return nil
+	}
+	return m.States.Delete(StateApplied, key)
+}
@@ -0,0 +1,66 @@
+package main
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+)
+
+type memStates map[string]json.RawMessage
+
+func (s memStates) Put(state, key string, v any) error {
+	b, _ := json.Marshal(v)
+	s[state+"/"+key] = b
+	return nil
+}
+func (s memStates) Delete(state, key string) error { delete(s, state+"/"+key); return nil }
+func (s memStates) Get(state, key string) (json.RawMessage, bool, error) {
+	b, ok := s[state+"/"+key]
+	return b, ok, nil
+}
+
+// §25, §26: each change emits its event with before and after and keeps its key of the state `applied`, a
+// refusal and a wait too, a wait keeping its first since, and a value back to not managed leaves the state.
+func TestEachChangeIsSaidAndKeptAndANotManagedValueLeavesTheState(t *testing.T) {
+	f := newFake(t)
+	m := f.machine()
+	st := memStates{}
+	m.States = st
+	if err := m.Changed("pool", "pool-properties.autotrim", "off", "on", "on"); err != nil {
+		t.Fatal(err)
+	}
+	e := f.emitted[0]
+	if e["_type"] != "config.applied" || e["target"] != "pool" || e["key"] != "pool-properties.autotrim" ||
+		e["before"] != "off" || e["after"] != "on" || e["read-back"] != "on" {
+		t.Errorf("event: %v", e)
+	}
+	var got AppliedEntry
+	json.Unmarshal(st["applied/pool-properties.autotrim"], &got)
+	if got.Before != "off" || got.After != "on" || got.Outcome != OutcomeApplied || !got.At.Equal(f.now.UTC()) {
+		t.Errorf("state: %+v", got)
+	}
+	if err := m.Refused(Refusal{Key: "hostid", Value: "zzz", Reason: "is not eight hexadecimal digits"}); err != nil {
+		t.Fatal(err)
+	}
+	if f.emitted[1]["_type"] != "config.refused" || f.emitted[1]["reason"] == "" || !strings.Contains(string(st["applied/hostid"]), `"refused"`) {
+		t.Errorf("refusal: %v %s", f.emitted[1], st["applied/hostid"])
+	}
+	m.Waiting("log-devices", "the pool's scrub")
+	first := f.now
+	f.now = f.now.Add(time.Hour)
+	m.Waiting("log-devices", "the pool's scrub")
+	var w AppliedEntry
+	json.Unmarshal(st["applied/log-devices"], &w)
+	if f.emitted[3]["_type"] != "config.waiting" || !w.Since.Equal(first.UTC()) || !w.At.Equal(f.now.UTC()) || w.Outcome != OutcomeWaiting {
+		t.Errorf("wait: %v %+v", f.emitted[3], w)
+	}
+	m.NotManaged("log-devices")
+	if _, ok := st["applied/log-devices"]; ok {
+		t.Error("the key stays after the value went back to not managed")
+	}
+	m.Emit = nil
+	if err := m.Changed("pool", "k", "a", "b", "b"); err == nil {
+		t.Error("an event that cannot be said is said as done")
+	}
+}
@@ -113,6 +113,8 @@ type Machine struct {
 	ReadFile func(path string) ([]byte, error)
 	// Emit publishes an event as this module (the record of an act); nil where nothing can be published.
 	Emit func(eventType string, body any) error
+	// States is the module's states on the bus (§26); nil where this process reaches none.
+	States StateStore
 	// Errors is what the health check last saw of each device's error counts; nil judges every count
 	// as past.
 	Errors *ErrorMemory