zfs: the tools say events with mesh/stream.publish and a hand act hears its event on the events stream (A-42, A-12)

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

@@ -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)
+		}
 	}
 }
 

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")
+	}
+}

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

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)
 		}

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

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)