From 1ff904f10107e4581b50c6a25761ba42602fd29a Mon Sep 17 00:00:00 2001 From: jochens Date: Fri, 13 Nov 2026 01:50:27 +0100 Subject: [PATCH] zfs: the tools say events with mesh/stream.publish and a hand act hears its event on the events stream (A-42, A-12) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tools say every event of §25 (snapshot.taken.person, scrub.requested and the others) with mesh/stream.publish on the stream events, never stdio.Emit; the hand acts' 30 s wait reads the mark (stream.fill) before the act starts, then stream.read from the next sequence, for at most HandWait. Builds: design 42 zfs-owns-zfs §25, §24, lead answers A-42, A-12 --- modules/zfs/cmd/zfs-tools/bus.go | 106 ++++++++++-------- modules/zfs/cmd/zfs-tools/bus_stream_test.go | 112 +++++++++++++++++++ modules/zfs/cmd/zfs-tools/hand.go | 8 +- modules/zfs/cmd/zfs-tools/hand_test.go | 15 ++- modules/zfs/cmd/zfs-tools/machine.go | 6 +- modules/zfs/cmd/zfs-tools/main.go | 5 +- 6 files changed, 197 insertions(+), 55 deletions(-) create mode 100644 modules/zfs/cmd/zfs-tools/bus_stream_test.go diff --git a/modules/zfs/cmd/zfs-tools/bus.go b/modules/zfs/cmd/zfs-tools/bus.go index b5da08576..d1f072b82 100644 --- a/modules/zfs/cmd/zfs-tools/bus.go +++ b/modules/zfs/cmd/zfs-tools/bus.go @@ -134,55 +134,73 @@ func (toolStates) Get(state, key string) (json.RawMessage, bool, error) { 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{} +// toolEmit says one event of the module on its own events stream, by the tool runner's mesh/stream.publish (A-42): the +// runner publishes it acknowledged on the stream's subject plus the kind, the id its message id. Never stdio.Emit, which +// publishes on the module's old event subjects. The id (nanotime-pid) is the time in nanoseconds and the process id, so +// two events of one process never share one. +func toolEmit(ask func(method string, params any) (json.RawMessage, error)) func(kind string, body any) error { + return func(kind string, body any) error { + raw, err := json.Marshal(body) + if err != nil { + return err + } + _, err = ask(buscalls.StreamPublish, buscalls.StreamPublishRequest{Stream: EventsStream, Event: kind, Body: raw, + ID: fmt.Sprintf("%d-%d", time.Now().UnixNano(), os.Getpid())}) + return err } - 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: +} + +// 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 + } } } - return nil - }) - } - select { - case ev := <-w.ch: - return ev, true - case <-time.After(bound): - return nil, false + if !h.now().Before(deadline) { + return nil, false + } + h.sleep(HandPoll) + } } } diff --git a/modules/zfs/cmd/zfs-tools/bus_stream_test.go b/modules/zfs/cmd/zfs-tools/bus_stream_test.go new file mode 100644 index 000000000..983b348ec --- /dev/null +++ b/modules/zfs/cmd/zfs-tools/bus_stream_test.go @@ -0,0 +1,112 @@ +package main + +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 + var method string + say := toolEmit(func(m string, p any) (json.RawMessage, error) { + method, got = m, p.(buscalls.StreamPublishRequest) + return json.RawMessage(`{"sequence":9}`), nil + }) + if err := say(ScrubEvent, map[string]any{"pool": "tank"}); err != nil { + t.Fatal(err) + } + if method != buscalls.StreamPublish || got.Stream != "events" || got.Event != ScrubEvent || got.ID == "" || string(got.Body) != `{"pool":"tank"}` { + t.Errorf("asked %s %+v", method, got) + } + fail := toolEmit(func(string, any) (json.RawMessage, error) { return nil, errors.New("refused") }) + if fail(ScrubEvent, nil) == nil { + t.Error("a refused publish was said done") + } +} diff --git a/modules/zfs/cmd/zfs-tools/hand.go b/modules/zfs/cmd/zfs-tools/hand.go index a85c28704..6d30d4c83 100644 --- a/modules/zfs/cmd/zfs-tools/hand.go +++ b/modules/zfs/cmd/zfs-tools/hand.go @@ -66,6 +66,10 @@ 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) } @@ -73,8 +77,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 m.Await != nil { - if ev, ok := m.Await(handEnds[act], HandWait); ok { + 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 diff --git a/modules/zfs/cmd/zfs-tools/hand_test.go b/modules/zfs/cmd/zfs-tools/hand_test.go index c4f685e86..39bf1344e 100644 --- a/modules/zfs/cmd/zfs-tools/hand_test.go +++ b/modules/zfs/cmd/zfs-tools/hand_test.go @@ -37,12 +37,16 @@ func TestAHandActWritesItsKeyStartsTheRunAndAnswersFromItsEvent(t *testing.T) { 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 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, d time.Duration) (map[string]any, bool) { - kinds, bound = k, d - return c.event, c.event != nil + 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 + } } a, err := m.HandAct(context.Background(), c.act, "caller@node", c.why) if c.err != "" { @@ -61,6 +65,9 @@ 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) } diff --git a/modules/zfs/cmd/zfs-tools/machine.go b/modules/zfs/cmd/zfs-tools/machine.go index 55ad575b1..fd2e6e0c5 100644 --- a/modules/zfs/cmd/zfs-tools/machine.go +++ b/modules/zfs/cmd/zfs-tools/machine.go @@ -125,9 +125,9 @@ 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 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) + // 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 diff --git a/modules/zfs/cmd/zfs-tools/main.go b/modules/zfs/cmd/zfs-tools/main.go index b10224215..bfa579340 100644 --- a/modules/zfs/cmd/zfs-tools/main.go +++ b/modules/zfs/cmd/zfs-tools/main.go @@ -16,6 +16,7 @@ import ( "os" "os/signal" "syscall" + "time" stdio "git.novox.be/novox/mesh-sdk/go" ) @@ -71,8 +72,8 @@ func main() { return } m := Here() - m.Emit = func(eventType string, body any) error { return stdio.Emit(eventType, body) } - m.States, m.Call, m.Ledger, m.Await, m.Log = toolStates{}, stdio.Ask, busLedger{}, (&eventHub{}).await, busStream{} + 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{} // 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)