novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

zfs: the hand acts answer started at once and the manifest asks stream-publish, bus-state-calls and ledger-calls (A-43, A-44)

Browse Source 
Removes streamHear, HandPoll, HandWait and the stream mark; keeps toolEmit.

Builds: design 42 zfs-owns-zfs §24, §25, lead answers A-43, A-44
This commit is contained in:
jschoubben
2026-11-13 01:52:50 +01:00
parent 1ff904f101
commit 58664eee60
8 changed files with 26 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/zfs/cmd/zfs-tools/bus.go
-54
View File
@@ -150,60 +150,6 @@ func toolEmit(ask func(method string, params any) (json.RawMessage, error)) func
}
}
// HandPoll is how long a hand act's wait lets pass between two reads of the events stream.
const HandPoll = 500 * time.Millisecond
// streamHear lets a hand act hear its act's event by reading the module's own events stream with the calls a bundle has
// (A-12, A-42): the stream's last sequence is read before the act starts, then the stream from the next sequence on.
type streamHear struct {
bus func(method string, params any) (json.RawMessage, error)
now func() time.Time
sleep func(time.Duration)
}
// await marks the stream's last sequence and answers the wait for an event of one of kinds after it. A stream that
// cannot be read answers a wait that sees nothing, so the act is answered "started".
func (h streamHear) await(kinds []string) func(bound time.Duration) (map[string]any, bool) {
raw, err := h.bus(buscalls.StreamFill, buscalls.StreamRequest{Stream: EventsStream})
var fill buscalls.StreamFilled
if err == nil {
err = json.Unmarshal(raw, &fill)
}
if err != nil {
return func(time.Duration) (map[string]any, bool) { return nil, false }
}
from := fill.Last + 1
return func(bound time.Duration) (map[string]any, bool) {
deadline := h.now().Add(bound)
for {
raw, err := h.bus(buscalls.StreamRead, buscalls.StreamReadRequest{Stream: EventsStream, From: from, Limit: 500})
if err == nil {
var page buscalls.StreamPage
if json.Unmarshal(raw, &page) == nil {
for _, msg := range page.Messages {
from = msg.Sequence + 1
for _, k := range kinds {
if msg.Subject == k || strings.HasSuffix(msg.Subject, "."+k) {
var body map[string]any
_ = json.Unmarshal(msg.Body, &body)
return body, true
}
}
}
if page.Next != 0 {
from = page.Next
continue
}
}
}
if !h.now().Before(deadline) {
return nil, false
}
h.sleep(HandPoll)
}
}
}
// busLedger is the ledger `snapshots` through the tool runner.
type busLedger struct{}
modules/zfs/cmd/zfs-tools/bus_stream_test.go
-83
View File
@@ -4,93 +4,10 @@ import (
"encoding/json"
"errors"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/buscalls"
)
// fakeStream answers a bundle's calls on its events stream from a list of messages and a clock the wait advances.
type fakeStream struct {
msgs []buscalls.StreamMessage
asked []string
reads []uint64
now time.Time
fillErr error
// at maps a read count to messages added before that read (an event that comes while the tool waits).
arrive map[int]buscalls.StreamMessage
}
func (f *fakeStream) ask(method string, params any) (json.RawMessage, error) {
f.asked = append(f.asked, method)
switch method {
case buscalls.StreamFill:
if f.fillErr != nil {
return nil, f.fillErr
}
last := uint64(0)
if n := len(f.msgs); n > 0 {
last = f.msgs[n-1].Sequence
}
return json.Marshal(buscalls.StreamFilled{First: 1, Last: last})
case buscalls.StreamRead:
r := params.(buscalls.StreamReadRequest)
f.reads = append(f.reads, r.From)
if m, ok := f.arrive[len(f.reads)]; ok {
m.Sequence = uint64(len(f.msgs) + 1)
f.msgs = append(f.msgs, m)
}
page := buscalls.StreamPage{Messages: []buscalls.StreamMessage{}}
for _, m := range f.msgs {
if m.Sequence >= r.From {
page.Messages = append(page.Messages, m)
}
}
return json.Marshal(page)
case buscalls.StreamPublish:
return json.Marshal(buscalls.StreamPublished{Sequence: 9})
}
return nil, errors.New("unexpected " + method)
}
func (f *fakeStream) hear() streamHear {
return streamHear{bus: f.ask, now: func() time.Time { return f.now }, sleep: func(d time.Duration) { f.now = f.now.Add(d) }}
}
// A-12, A-42: a hand act's wait marks the events stream's last sequence before the act starts, then reads from the next
// one: an event before the mark is not the act's, one after it with an end kind is, and with none in 30 s the wait gives up.
func TestAHandWaitHearsTheEventsStreamAfterItsMark(t *testing.T) {
old := buscalls.StreamMessage{Sequence: 1, Subject: "mesh.bus.zfs.events.snapshot.taken.person", Body: json.RawMessage(`{"old":true}`)}
other := buscalls.StreamMessage{Subject: "mesh.bus.zfs.events.config.refused", Body: json.RawMessage(`{}`)}
end := buscalls.StreamMessage{Subject: "mesh.bus.zfs.events.snapshot.taken.person", Body: json.RawMessage(`{"snapshot":"tank@x"}`)}
f := &fakeStream{msgs: []buscalls.StreamMessage{old}, now: time.Unix(0, 0), arrive: map[int]buscalls.StreamMessage{2: other, 3: end}}
wait := f.hear().await([]string{EventTakenPerson})
if len(f.asked) != 1 || f.asked[0] != buscalls.StreamFill {
t.Fatalf("the mark was not read first: %v", f.asked)
}
ev, ok := wait(HandWait)
if !ok || ev["snapshot"] != "tank@x" {
t.Fatalf("heard %v %v; the old event or none", ev, ok)
}
if f.reads[0] != 2 {
t.Errorf("read from %d, want the sequence after the mark, 2", f.reads[0])
}
g := &fakeStream{msgs: []buscalls.StreamMessage{old}, now: time.Unix(0, 0)}
start := g.now
if ev, ok := g.hear().await([]string{EventTakenPerson})(HandWait); ok || ev != nil {
t.Fatalf("heard %v with nothing new on the stream", ev)
}
if g.now.Sub(start) < HandWait {
t.Errorf("gave up after %s, before the 30 s", g.now.Sub(start))
}
h := &fakeStream{fillErr: errors.New("no stream"), now: time.Unix(0, 0)}
if _, ok := h.hear().await([]string{EventTakenPerson})(HandWait); ok {
t.Error("heard an event from a stream that could not be read")
}
}
// A-42: the tool says an event with mesh/stream.publish on the stream `events`, the kind as the event and a message id.
func TestAToolSaysItsEventWithStreamPublish(t *testing.T) {
var got buscalls.StreamPublishRequest
modules/zfs/cmd/zfs-tools/hand.go
+10 -30
View File
@@ -4,19 +4,15 @@ 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
// acts under the same lock and guards, and deletes it. The tool does not wait (A-43): once the key is written and
// the unit started it answers "started" at once, 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.
// The acts, as the keys of the state `schedule`.
const (
@@ -25,12 +21,8 @@ const (
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},
}
// handActs are the acts of the run a person may start.
var handActs = map[string]bool{ActSnapshots: true, ActPrune: true, ActApply: true}
// 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} }
@@ -38,10 +30,9 @@ func StartArgs() map[string]any { return map[string]any{"scope": "system", "unit
// 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"`
// Outcome is "started": the act was started and its end is read from zfs_module_events (A-43).
Outcome string `json:"outcome"`
Said string `json:"said"`
}
func whyOf(why string) error {
@@ -57,7 +48,7 @@ func whyOf(why string) error {
// 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 {
if !handActs[act] {
return a, fmt.Errorf("%q is not an act of the run: snapshots, prune or apply", act)
}
if err := whyOf(why); err != nil {
@@ -66,10 +57,6 @@ func (m *Machine) HandAct(ctx context.Context, act, by, why string) (HandAnswer,
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)
}
var wait func(time.Duration) (map[string]any, bool)
if m.Await != nil {
wait = m.Await(handEnds[act]) // marks the stream's last sequence before the act starts
}
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)
}
@@ -77,15 +64,8 @@ func (m *Machine) HandAct(ctx context.Context, act, by, why string) (HandAnswer,
_ = m.States.Delete(StateSchedule, act)
return a, fmt.Errorf("the service manager did not start %s: %v", RunService, err)
}
if wait != nil {
if ev, ok := wait(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)
a.Said = fmt.Sprintf("Started the act %s (%s) and did not wait for it to end: zfs_module_events reads its outcome.", act, RunService)
return a, nil
}
modules/zfs/cmd/zfs-tools/hand_test.go
+8 -25
View File
@@ -6,26 +6,22 @@ import (
"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
// answers "started" at once and never "done" (A-43); a failed start
// leaves no key; pause writes `paused`, resume deletes it.
func TestAHandActWritesItsKeyStartsTheRunAndAnswersFromItsEvent(t *testing.T) {
func TestAHandActWritesItsKeyStartsTheRunAndAnswersStartedAtOnce(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: "snapshots", act: ActSnapshots, why: "before the move", outcome: "started", keyLeft: true},
{name: "prune", act: ActPrune, why: "free space", outcome: "started", keyLeft: true},
{name: "apply", act: ActApply, why: "now", outcome: "started", keyLeft: true},
{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"},
@@ -37,17 +33,7 @@ func TestAHandActWritesItsKeyStartsTheRunAndAnswersFromItsEvent(t *testing.T) {
m.States, m.Node = st, "ace"
var key string
var body any
var marked, started bool
m.Call = func(k string, b any) (json.RawMessage, error) { key, body, started = k, b, true; return nil, c.callErr }
var bound time.Duration
var kinds []string
m.Await = func(k []string) func(time.Duration) (map[string]any, bool) {
kinds, marked = k, !started
return func(d time.Duration) (map[string]any, bool) {
bound = d
return c.event, c.event != nil
}
}
m.Call = func(k string, b any) (json.RawMessage, error) { key, body = k, b; return nil, c.callErr }
a, err := m.HandAct(context.Background(), c.act, "caller@node", c.why)
if c.err != "" {
if err == nil || !strings.Contains(err.Error(), c.err) {
@@ -65,11 +51,8 @@ func TestAHandActWritesItsKeyStartsTheRunAndAnswersFromItsEvent(t *testing.T) {
t.Errorf("%s: called %s with %v", c.name, key, body)
}
}
if c.err == "" && !marked {
t.Errorf("%s: the stream was not marked before the act started", c.name)
}
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, "zfs_module_events") {
t.Errorf("%s: did not say where the outcome is read: %s", c.name, a.Said)
}
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)
modules/zfs/cmd/zfs-tools/machine.go
-3
View File
@@ -125,9 +125,6 @@ type Machine struct {
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 marks the module's events stream before an act starts and answers the wait for an event of the module whose
// kind is one of kinds, after the mark: at most the bound, and the event; false when none came (A-12, A-42).
Await func(kinds []string) func(bound time.Duration) (map[string]any, bool)
// Findings writes the module's own state `findings` as the seat mesh-conditions' layout says (A-32).
Findings views.FindingWriter
// Warnings are what the apply saw that is no failure (a change the bus did not take, a parameter awaiting
modules/zfs/cmd/zfs-tools/main.go
+1 -2
View File
@@ -16,7 +16,6 @@ import (
"os"
"os/signal"
"syscall"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
@@ -73,7 +72,7 @@ func main() {
}
m := Here()
m.Emit = toolEmit(stdio.BusState)
m.States, m.Call, m.Ledger, m.Await, m.Log = toolStates{}, stdio.Ask, busLedger{}, streamHear{bus: stdio.BusState, now: time.Now, sleep: time.Sleep}.await, busStream{}
m.States, m.Call, m.Ledger, m.Log = toolStates{}, stdio.Ask, busLedger{}, busStream{}
// 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/tools.go
+3 -3
View File
@@ -62,9 +62,9 @@ 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)"),
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. Does not wait: answers started at once 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). Does not wait: answers started at once 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. Does not wait: answers started at once 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)",
modules/zfs/module.json
+4 -1
View File
@@ -2,8 +2,11 @@
"module": "zfs",
"version": "1",
"capabilities": [
"bus-state-calls",
"ledger-calls",
"package-manager",
"service-manager"
"service-manager",
"stream-publish"
],
"invokes": [
"seat:node-service-manager.start"