From a775b42e4d6a88e30cdc90d1ef9d36988b7fa54b Mon Sep 17 00:00:00 2001 From: jochens Date: Fri, 13 Nov 2026 01:49:02 +0100 Subject: [PATCH 1/2] tool runner: mesh/stream.publish says an event on the calling bundle's own bound stream (A-42) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 42 zfs-owns-zfs §25, §24, lead answer A-42 --- node-tools/internal/bus/bindings.go | 35 +++++++++++++++++++ node-tools/internal/bus/bindings_test.go | 26 ++++++++++++++ node-tools/internal/launch/launch.go | 2 +- .../internal/launch/stream_publish_test.go | 14 ++++++++ node-tools/internal/runtime/busstate.go | 7 +++- .../internal/runtime/stream_publish_test.go | 15 ++++++++ 6 files changed, 97 insertions(+), 2 deletions(-) create mode 100644 node-tools/internal/launch/stream_publish_test.go create mode 100644 node-tools/internal/runtime/stream_publish_test.go diff --git a/node-tools/internal/bus/bindings.go b/node-tools/internal/bus/bindings.go index 8937a61..26cd793 100644 --- a/node-tools/internal/bus/bindings.go +++ b/node-tools/internal/bus/bindings.go @@ -275,6 +275,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 diff --git a/node-tools/internal/bus/bindings_test.go b/node-tools/internal/bus/bindings_test.go index 81c25a0..7cb1f25 100644 --- a/node-tools/internal/bus/bindings_test.go +++ b/node-tools/internal/bus/bindings_test.go @@ -2,6 +2,7 @@ package bus_test import ( "context" + "encoding/json" "errors" "strings" "testing" @@ -81,6 +82,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 @@ -97,6 +115,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 diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index d293a19..5316862 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -120,7 +120,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, diff --git a/node-tools/internal/launch/stream_publish_test.go b/node-tools/internal/launch/stream_publish_test.go new file mode 100644 index 0000000..7a44e6e --- /dev/null +++ b/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") + } +} diff --git a/node-tools/internal/runtime/busstate.go b/node-tools/internal/runtime/busstate.go index 2c67519..624f13e 100644 --- a/node-tools/internal/runtime/busstate.go +++ b/node-tools/internal/runtime/busstate.go @@ -21,7 +21,7 @@ const ( // 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}, + buscalls.ObjectsWatch, buscalls.StreamRead, buscalls.StreamFill, buscalls.StreamPublish, buscalls.WorkQueueSize, buscalls.WorkQueueDead}, CapLedgerCalls: {buscalls.LedgerApply, buscalls.LedgerGet, buscalls.LedgerHistory, buscalls.LedgerWatch, buscalls.LedgerQuery, buscalls.LedgerRebuild, buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag}, } @@ -85,6 +85,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 { diff --git a/node-tools/internal/runtime/stream_publish_test.go b/node-tools/internal/runtime/stream_publish_test.go new file mode 100644 index 0000000..130c566 --- /dev/null +++ b/node-tools/internal/runtime/stream_publish_test.go @@ -0,0 +1,15 @@ +package runtime + +import ( + "slices" + "testing" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// A-42: mesh/stream.publish is part of the bus-state-calls capability a module's manifest asks for. +func TestStreamPublishIsPartOfTheBusStateCallsCapability(t *testing.T) { + if !slices.Contains(callFamilies[CapBusStateCalls], buscalls.StreamPublish) { + t.Fatal("bus-state-calls does not state mesh/stream.publish") + } +} -- 2.54.0 From df2162e381f5c192884e81dbb49de782ac6ca9da Mon Sep 17 00:00:00 2001 From: jochens Date: Fri, 13 Nov 2026 01:52:15 +0100 Subject: [PATCH 2/2] tool runner: mesh/stream.publish is the capability stream-publish, taken out of bus-state-calls (A-44) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 42 zfs-owns-zfs §25, lead answer A-44 --- node-tools/internal/runtime/announce_test.go | 2 +- node-tools/internal/runtime/busstate.go | 5 ++++- node-tools/internal/runtime/ledgers_test.go | 2 +- node-tools/internal/runtime/stream_publish_test.go | 11 +++++++---- 4 files changed, 13 insertions(+), 7 deletions(-) diff --git a/node-tools/internal/runtime/announce_test.go b/node-tools/internal/runtime/announce_test.go index 6965fea..2896130 100644 --- a/node-tools/internal/runtime/announce_test.go +++ b/node-tools/internal/runtime/announce_test.go @@ -139,7 +139,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] != "bus-state-calls,ledger-calls,seat-settle-delay" { + if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls,seat-settle-delay,stream-publish" { t.Errorf("the runner states %q (%v), want its bus-state and ledger calls", ping.Metadata[announce.Capabilities], err) } } diff --git a/node-tools/internal/runtime/busstate.go b/node-tools/internal/runtime/busstate.go index 624f13e..fd43409 100644 --- a/node-tools/internal/runtime/busstate.go +++ b/node-tools/internal/runtime/busstate.go @@ -16,12 +16,15 @@ import ( const ( CapBusStateCalls = "bus-state-calls" CapLedgerCalls = "ledger-calls" + // CapStreamPublish is the call that says an event on the module's own bound stream (A-44). + CapStreamPublish = "stream-publish" ) // 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.StreamPublish, buscalls.WorkQueueSize, buscalls.WorkQueueDead}, + 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}, } diff --git a/node-tools/internal/runtime/ledgers_test.go b/node-tools/internal/runtime/ledgers_test.go index 805984a..23bf5fb 100644 --- a/node-tools/internal/runtime/ledgers_test.go +++ b/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 != "bus-state-calls,ledger-calls,seat-settle-delay" { + if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls,seat-settle-delay,stream-publish" { t.Fatalf("the runner states %s", got) } for capability, methods := range callFamilies { diff --git a/node-tools/internal/runtime/stream_publish_test.go b/node-tools/internal/runtime/stream_publish_test.go index 130c566..0bc66ff 100644 --- a/node-tools/internal/runtime/stream_publish_test.go +++ b/node-tools/internal/runtime/stream_publish_test.go @@ -7,9 +7,12 @@ import ( "git.novox.be/novox/mesh-sdk/go/buscalls" ) -// A-42: mesh/stream.publish is part of the bus-state-calls capability a module's manifest asks for. -func TestStreamPublishIsPartOfTheBusStateCallsCapability(t *testing.T) { - if !slices.Contains(callFamilies[CapBusStateCalls], buscalls.StreamPublish) { - t.Fatal("bus-state-calls does not state mesh/stream.publish") +// 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") } } -- 2.54.0