diff --git a/node-tools/internal/bus/bindings.go b/node-tools/internal/bus/bindings.go index 3a441bf..85ead67 100644 --- a/node-tools/internal/bus/bindings.go +++ b/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 diff --git a/node-tools/internal/bus/bindings_test.go b/node-tools/internal/bus/bindings_test.go index d6409a4..8e163d6 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" "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 diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index 0dbe175..203b28e 100644 --- a/node-tools/internal/launch/launch.go +++ b/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, 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/announce_test.go b/node-tools/internal/runtime/announce_test.go index 43556e6..6999ec6 100644 --- a/node-tools/internal/runtime/announce_test.go +++ b/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 { diff --git a/node-tools/internal/runtime/busstate.go b/node-tools/internal/runtime/busstate.go index 8f9a3a2..567c3de 100644 --- a/node-tools/internal/runtime/busstate.go +++ b/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 { diff --git a/node-tools/internal/runtime/ledgers_test.go b/node-tools/internal/runtime/ledgers_test.go index 3b58606..ff586ef 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 != "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 { 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..0bc66ff --- /dev/null +++ b/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") + } +} diff --git a/node-tools/module.json b/node-tools/module.json index aeb6876..ebdfa46 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -67,7 +67,8 @@ "ledger-upkeep-calls", "put-sessions", "seat-settle-delay", - "sent-objects-calls" + "sent-objects-calls", + "stream-publish" ] } ]