Merge main into feat/47-G-join11: the tool runner's mesh/stream.publish (issue 1979)

The capability stream-publish is stated under mesh-sdk's word (buscalls.StreamPublishWord) and in the runtime artifact's states, so G's test that the stated words are the manifest's states holds; the publish uses G's exported JetStream.

Builds: design 47 G §545, §622; design 42 zfs-owns-zfs A-42, A-44

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

node-tools/internal/bus/bindings.go

@@ -276,6 +276,41 @@ func (c *Conn) StreamFill(module string, r buscalls.StreamRequest) (buscalls.Str
 	return c.streamFill(b.Stream)
 }
 
+// StreamPublish says one event on the module's own bound stream, by an acknowledged publish on the binding's subject plus
+// "." plus the event, the id as the message id (design 42 zfs-owns-zfs, A-42). The module's own binding alone: another
+// module's, and a binding that is not a stream, are refused by Binding.
+func (c *Conn) StreamPublish(module string, r buscalls.StreamPublishRequest) (buscalls.StreamPublished, error) {
+	b, err := c.streamBinding(module, r.Stream)
+	if err != nil {
+		return buscalls.StreamPublished{}, err
+	}
+	if b.Subject == "" {
+		return buscalls.StreamPublished{}, fmt.Errorf("%s's stream %q has no subject to say an event on", module, r.Stream)
+	}
+	if r.Event == "" || strings.ContainsAny(r.Event, " *>\t\r\n") || strings.HasPrefix(r.Event, ".") || strings.HasSuffix(r.Event, ".") {
+		return buscalls.StreamPublished{}, fmt.Errorf("the event %q is not a name to say on %s's stream %q", r.Event, module, r.Stream)
+	}
+	body := []byte(r.Body)
+	if len(body) == 0 {
+		body = []byte("null")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	js, err := c.JetStream()
+	if err != nil {
+		return buscalls.StreamPublished{}, err
+	}
+	var opts []jetstream.PublishOpt
+	if r.ID != "" {
+		opts = append(opts, jetstream.WithMsgID(r.ID))
+	}
+	ack, err := js.Publish(ctx, b.Subject+"."+r.Event, body, opts...)
+	if err != nil {
+		return buscalls.StreamPublished{}, fmt.Errorf("the event %s was not acknowledged on %s's stream %q: %w", r.Event, module, r.Stream, err)
+	}
+	return buscalls.StreamPublished{Sequence: ack.Sequence}, nil
+}
+
 // StreamReadMost is the most messages one read answers.
 const StreamReadMost = 1000
 

node-tools/internal/bus/bindings_test.go

@@ -2,6 +2,7 @@ package bus_test
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"io"
 	"strings"
@@ -99,6 +100,23 @@ func TestABundleReachesItsOwnBoundObjectsAndStreamsOnly(t *testing.T) {
 		t.Fatalf("fill %+v, %v", fill, err)
 	}
 
+	// A-42: an event said on the module's own stream lands on the binding's subject plus the event, acknowledged, and a
+	// second say with the same id is the same message.
+	said, err := conn.StreamPublish("m", buscalls.StreamPublishRequest{Stream: "events", Event: "scrub.requested",
+		Body: json.RawMessage(`{"pool":"tank"}`), ID: "id-1"})
+	if err != nil || said.Sequence != 3 {
+		t.Fatalf("said %+v, %v", said, err)
+	}
+	if again, err := conn.StreamPublish("m", buscalls.StreamPublishRequest{Stream: "events", Event: "scrub.requested",
+		Body: json.RawMessage(`{"pool":"tank"}`), ID: "id-1"}); err != nil || again.Sequence != 3 {
+		t.Fatalf("said again %+v, %v", again, err)
+	}
+	if page, err := conn.StreamRead("m", buscalls.StreamReadRequest{Stream: "events", From: 3,
+		Subjects: "mesh.state.m.events.scrub.requested"}); err != nil || len(page.Messages) != 1 ||
+		string(page.Messages[0].Body) != `{"pool":"tank"}` {
+		t.Fatalf("read back %+v, %v", page, err)
+	}
+
 	for _, c := range []struct {
 		name, module, refusal string
 		call                  func(module string) error
@@ -115,6 +133,14 @@ func TestABundleReachesItsOwnBoundObjectsAndStreamsOnly(t *testing.T) {
 			_, err := conn.ObjectsList(module, buscalls.ObjectsRequest{Objects: "events"})
 			return err
 		}},
+		{"say on another module's stream", "other", `other was bound no stream called "events"`, func(module string) error {
+			_, err := conn.StreamPublish(module, buscalls.StreamPublishRequest{Stream: "events", Event: "x"})
+			return err
+		}},
+		{"say on a binding that is not a stream", "m", `"files" is bound with the storage type object-store, not stream`, func(module string) error {
+			_, err := conn.StreamPublish(module, buscalls.StreamPublishRequest{Stream: "files", Event: "x"})
+			return err
+		}},
 		{"a work queue not bound", "m", `bound no stream called "jobs"`, func(module string) error {
 			_, err := conn.WorkQueueSize(module, buscalls.WorkQueueRequest{Queue: "jobs"})
 			return err

node-tools/internal/launch/launch.go

@@ -121,7 +121,7 @@ type Bus interface {
 // busStateCalls are the methods of mesh-sdk buscalls a bundle asks and BusState answers.
 var busStateCalls = map[string]bool{
 	buscalls.ObjectsPut: true, buscalls.ObjectsGet: true, buscalls.ObjectsList: true, buscalls.ObjectsDelete: true,
-	buscalls.ObjectsInfo: true, buscalls.StreamRead: true, buscalls.StreamFill: true, buscalls.WorkQueueSize: true,
+	buscalls.ObjectsInfo: true, buscalls.StreamRead: true, buscalls.StreamFill: true, buscalls.StreamPublish: true, buscalls.WorkQueueSize: true,
 	buscalls.WorkQueueDead: true, buscalls.LedgerApply: true, buscalls.LedgerGet: true, buscalls.LedgerHistory: true,
 	buscalls.LedgerQuery: true, buscalls.LedgerRebuild: true, buscalls.LedgerReindex: true, buscalls.LedgerSnapshot: true,
 	buscalls.LedgerLag: true, buscalls.ObjectsSentGet: true, buscalls.ObjectsPutChunk: true, buscalls.ObjectsGetChunk: true,

node-tools/internal/launch/stream_publish_test.go

@@ -0,0 +1,14 @@
+package launch
+
+import (
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+)
+
+// A-42: a launched bundle may ask mesh/stream.publish.
+func TestABundleMayAskStreamPublish(t *testing.T) {
+	if !busStateCalls[buscalls.StreamPublish] {
+		t.Fatal("mesh/stream.publish is not among the methods a bundle may ask")
+	}
+}

node-tools/internal/runtime/announce_test.go

@@ -144,7 +144,7 @@ func TestTheRunnerStatesTheCapabilitiesOfItsCallsInItsPing(t *testing.T) {
 		t.Fatalf("ping of the instance: %v", err)
 	}
 	var ping micro.Ping
-	if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "approval-mark/1,bus-state-calls,finding-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" {
+	if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "approval-mark/1,bus-state-calls,finding-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls,stream-publish" {
 		t.Errorf("the runner states %q (%v), want its bus-state and ledger calls", ping.Metadata[announce.Capabilities], err)
 	}
 	// Its own fingerprint beside them (design 46 approval mark §6).
@@ -161,6 +161,7 @@ func TestTheRunnerStatesTheCapabilitiesOfItsCallsInItsPing(t *testing.T) {
 			buscalls.LedgerRemove:   buscalls.LedgerUpkeepCalls,
 			buscalls.LedgerApply:    buscalls.LedgerCalls,
 			buscalls.ObjectsPut:     buscalls.BusStateCalls,
+			buscalls.StreamPublish:  buscalls.StreamPublishWord,
 		} {
 			var in []string
 			for c, methods := range callFamilies {

node-tools/internal/runtime/busstate.go

@@ -29,12 +29,16 @@ const (
 	// CapApprovalMark says the runner honours the approval mark: its gate sits on every handler it serves (design 46
 	// 7-approval-mark §6, §20), so the grants part composes no deny of its modules' marked subjects.
 	CapApprovalMark = "approval-mark/1"
+	// CapStreamPublish is the call that says an event on the module's own bound stream (design 42 zfs-owns-zfs, A-44), the
+	// word mesh-sdk's buscalls names (§622).
+	CapStreamPublish = buscalls.StreamPublishWord
 )
 
 // callFamilies are the methods each capability stands for, the ones BusState and BusStateWatch answer.
 var callFamilies = map[string][]string{
 	CapBusStateCalls: {buscalls.ObjectsPut, buscalls.ObjectsGet, buscalls.ObjectsList, buscalls.ObjectsDelete, buscalls.ObjectsInfo,
 		buscalls.ObjectsWatch, buscalls.StreamRead, buscalls.StreamFill, buscalls.WorkQueueSize, buscalls.WorkQueueDead},
+	CapStreamPublish: {buscalls.StreamPublish},
 	CapLedgerCalls: {buscalls.LedgerApply, buscalls.LedgerGet, buscalls.LedgerHistory, buscalls.LedgerWatch, buscalls.LedgerQuery,
 		buscalls.LedgerRebuild, buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag},
 	CapLedgerUpkeepCalls: {buscalls.LedgerFill, buscalls.LedgerRemove},
@@ -110,6 +114,11 @@ func (b *moduleBus) BusState(method string, params json.RawMessage) (json.RawMes
 		if err = decode(method, params, &r); err == nil {
 			answer, err = conn.StreamFill(b.module, r)
 		}
+	case buscalls.StreamPublish:
+		var r buscalls.StreamPublishRequest
+		if err = decode(method, params, &r); err == nil {
+			answer, err = conn.StreamPublish(b.module, r)
+		}
 	case buscalls.WorkQueueSize:
 		var r buscalls.WorkQueueRequest
 		if err = decode(method, params, &r); err == nil {

node-tools/internal/runtime/ledgers_test.go

@@ -204,7 +204,7 @@ func TestABundleToolNamedAsALedgerRepairIsRefused(t *testing.T) {
 // no other name; none answers "the runtime answers no ...".
 func TestEveryCallOfAStatedCapabilityIsAnswered(t *testing.T) {
 	m, _, _ := aLedger(t)
-	if got := strings.Join(CallCapabilities(), ","); got != "approval-mark/1,bus-state-calls,finding-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" {
+	if got := strings.Join(CallCapabilities(), ","); got != "approval-mark/1,bus-state-calls,finding-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls,stream-publish" {
 		t.Fatalf("the runner states %s", got)
 	}
 	for capability, methods := range callFamilies {

node-tools/internal/runtime/stream_publish_test.go

@@ -0,0 +1,18 @@
+package runtime
+
+import (
+	"slices"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+)
+
+// A-44: mesh/stream.publish is a family of its own, the capability stream-publish, and no longer of bus-state-calls.
+func TestStreamPublishIsACapabilityOfItsOwn(t *testing.T) {
+	if !slices.Equal(callFamilies[CapStreamPublish], []string{buscalls.StreamPublish}) || CapStreamPublish != "stream-publish" {
+		t.Fatal("stream-publish does not state only mesh/stream.publish")
+	}
+	if slices.Contains(callFamilies[CapBusStateCalls], buscalls.StreamPublish) {
+		t.Fatal("bus-state-calls still states mesh/stream.publish")
+	}
+}

node-tools/module.json

@@ -67,7 +67,8 @@
           "ledger-upkeep-calls",
           "put-sessions",
           "seat-settle-delay",
-          "sent-objects-calls"
+          "sent-objects-calls",
+          "stream-publish"
         ]
       }
     ]