zfs: the keeper publishes events with an acknowledged JetStream publish on mesh.bus.zfs.events.<kind> (review finding 1, A-41)

Builds: design 42 zfs-owns-zfs A-41, §25

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

@@ -13,6 +13,8 @@ import (
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/buscalls"
 	"git.novox.be/novox/mesh-sdk/go/process"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
 )
 
 // busStates is the module's states as this process reaches them (its bus account's `state`).
@@ -73,10 +75,41 @@ func RunKeeper(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	k := &Keeper{Emit: func(kind string, body any) error { return c.Emit(kind, body, stdio.EmitOptions{}) }, States: busStates{c}, Call: c.Call}
+	_, js, err := acct.Dial()
+	if err != nil {
+		return err
+	}
+	k := &Keeper{Emit: func(kind string, body any) error { return publishEvent(js, acct, kind, body) }, States: busStates{c}, Call: c.Call}
 	return k.Serve(ctx, l)
 }
 
+// eventPublisher is the part of JetStream the keeper publishes with.
+type eventPublisher interface {
+	PublishMsg(ctx context.Context, m *nats.Msg, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)
+}
+
+// publishEvent puts one event on the module's events stream as a JetStream publish the bus acknowledges, on the
+// subject the stream captures, mesh.bus.<module>.events.<kind> (A-41); never process.Conn.Emit, whose subject the
+// keeper's account is not granted. The message id is a random one, as before.
+func publishEvent(js eventPublisher, acct process.Account, kind string, body any) error {
+	data, err := json.Marshal(body)
+	if err != nil {
+		return err
+	}
+	id := fmt.Sprintf("%d-%d", time.Now().UnixNano(), os.Getpid())
+	msg := nats.NewMsg(acct.BusSubject(EventsStream, kind))
+	msg.Header.Set("x-event-id", id)
+	msg.Header.Set("x-source", acct.Module)
+	msg.Header.Set("x-node", acct.Node)
+	msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
+	msg.Header.Set("content-type", "application/json")
+	msg.Data = data
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id))
+	return err
+}
+
 // toolStates is the module's states as the tools bundle reaches them, through the tool runner.
 type toolStates struct{}
 

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

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"context"
+	"testing"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/process"
+)
+
+type recordingPublisher struct{ msgs []*nats.Msg }
+
+func (r *recordingPublisher) PublishMsg(_ context.Context, m *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) {
+	r.msgs = append(r.msgs, m)
+	return &jetstream.PubAck{}, nil
+}
+
+// A-41, review finding 1: the keeper publishes on the subject the module's events stream captures.
+func TestTheKeeperPublishesOnTheEventsStreamsSubject(t *testing.T) {
+	p := &recordingPublisher{}
+	acct := process.Account{Node: "n1", Module: "zfs"}
+	if err := publishEvent(p, acct, EventApplied, map[string]string{"key": "x"}); err != nil {
+		t.Fatal(err)
+	}
+	if len(p.msgs) != 1 || p.msgs[0].Subject != "mesh.bus.zfs.events."+EventApplied || p.msgs[0].Header.Get("x-event-id") == "" {
+		t.Errorf("%+v", p.msgs)
+	}
+}

modules/zfs/go.mod

@@ -11,7 +11,7 @@ require (
 
 require (
 	github.com/klauspost/compress v1.20.1 // indirect
-	github.com/nats-io/nats.go v1.54.0 // indirect
+	github.com/nats-io/nats.go v1.54.0
 	github.com/nats-io/nkeys v0.4.16 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	golang.org/x/crypto v0.57.0 // indirect