diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 920189e..6b5419e 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -852,6 +852,12 @@ func answerOf(data []byte) (Answered, error) { // PublishAs emits an event as a module: published into JetStream and awaited, de-duplicated by its // own id (ADR 0042). The body is the payload; the metadata rides as headers. func (c *Conn) PublishAs(module string, env Envelope) error { + _, err := c.PublishAsAcked(module, env) + return err +} + +// PublishAsAcked is PublishAs that also answers the sequence the bus acknowledged the event at. +func (c *Conn) PublishAsAcked(module string, env Envelope) (uint64, error) { // On the module's own stream when its membership names one carrying the event (ADR 0363 §2), on EVENTS // otherwise — and only there, as before, for a membership that names no stream. msg := nats.NewMsg(eventSubject(c.Membership(module), module, env.Key)) @@ -873,8 +879,11 @@ func (c *Conn) PublishAs(module string, env Envelope) error { if id := env.Headers["x-event-id"]; id != "" { opts = append(opts, nats.MsgId(id)) } - _, err := c.js.PublishMsg(msg, opts...) - return err + ack, err := c.js.PublishMsg(msg, opts...) + if err != nil { + return 0, err + } + return ack.Sequence, nil } // Say publishes one message on core NATS, kept by nobody: a heartbeat, whose loss is the next one. diff --git a/node-tools/internal/bus/subscribenew.go b/node-tools/internal/bus/subscribenew.go new file mode 100644 index 0000000..d282fa3 --- /dev/null +++ b/node-tools/internal/bus/subscribenew.go @@ -0,0 +1,87 @@ +package bus + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "strings" + + "git.novox.be/novox/mesh-sdk/go/buscompose" + "github.com/nats-io/nats.go/jetstream" +) + +// SubscribeNew hears one event, named `.`, from the next message on, on an ordered consumer of its own on each +// stream that carries it, and hands every message to deliver as the envelope a module sees. The subjects the name lands on are +// mesh-sdk's one rule (buscompose.EventSubjects, design 47 G §559, A-153), so a name with a wildcard is refused, naming the rule; +// a name no stream on the bus carries is an error, never a subscription that hears nothing. stop ends every consumer. +func (c *Conn) SubscribeNew(name string, deliver func(json.RawMessage) error) (func(), error) { + subjects, err := buscompose.EventSubjects(name, "") + if err != nil { + return nil, err + } + js, err := c.JetStream() + if err != nil { + return nil, err + } + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + // The streams that carry each subject, found by the bus: a namespace no stream carries hears nothing and is skipped. + byStream := map[string][]string{} + for _, subject := range subjects { + if strings.HasSuffix(subject, ".>") { + continue + } + stream, err := js.StreamNameBySubject(ctx, subject) + if err != nil || stream == "" { + continue + } + byStream[stream] = append(byStream[stream], subject, subject+".>") + } + if len(byStream) == 0 { + return nil, fmt.Errorf("no stream on the bus carries %s (%s): nothing would be heard", name, strings.Join(subjects, ", ")) + } + var stops []func() + stopAll := func() { + for _, s := range stops { + s() + } + } + for _, stream := range slices.Sorted(mapKeys(byStream)) { + cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: byStream[stream], + DeliverPolicy: jetstream.DeliverNewPolicy}) + if err != nil { + stopAll() + return nil, fmt.Errorf("an ordered consumer of %s on %s: %w", name, stream, err) + } + cc, err := cons.Consume(func(m jetstream.Msg) { + env, ok := envelopeOf(m.Subject(), m.Headers(), m.Data()) + if !ok { + return + } + raw, err := json.Marshal(env) + if err != nil { + return + } + if err := deliver(raw); err != nil && c.Logf != nil { + c.Logf("[mesh-tools] a subscription for new %s messages did not take one: %v", name, err) + } + }) + if err != nil { + stopAll() + return nil, fmt.Errorf("consuming %s on %s: %w", name, stream, err) + } + stops = append(stops, cc.Stop) + } + return stopAll, nil +} + +func mapKeys(m map[string][]string) func(func(string) bool) { + return func(yield func(string) bool) { + for k := range m { + if !yield(k) { + return + } + } + } +} diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index ef3826f..0dbe175 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -387,7 +387,29 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an asked := time.Now() switch m.Method { case "mesh/publish": - err = mesh.Publish(m.Params) + var said struct { + Ack bool `json:"ack"` + } + _ = json.Unmarshal(m.Params, &said) + if !said.Ack { + err = mesh.Publish(m.Params) + break + } + // With ack the runtime answers only once the bus acknowledged the event: its sequence, or {error} + // (design 47 G §559). A refusal is an answer, so a writer says it as a finding's resend. + acker, ok := mesh.(interface { + PublishAcked(params json.RawMessage) (uint64, error) + }) + switch { + case !ok: + result = map[string]any{"error": "this tool runner's bus publishes no event with an acknowledgement"} + default: + if seq, perr := acker.PublishAcked(m.Params); perr != nil { + result = map[string]any{"error": perr.Error()} + } else { + result = map[string]any{"sequence": seq} + } + } case "mesh/ask": var asked json.RawMessage if asked, err = mesh.Ask(m.Params); err == nil { @@ -404,6 +426,38 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an result = json.RawMessage(raw) } case "mesh/subscribe": + var said struct { + Pattern string `json:"pattern"` + Deliver string `json:"deliver"` + } + _ = json.Unmarshal(m.Params, &said) + if said.Deliver == "new" { + // An ordered consumer of its own from the next message on (design 47 G §559, A-153): each message + // is handed to this child as an event is, and the consumer ends with the child. + newer, ok := mesh.(interface { + SubscribeNew(pattern string, deliver func(json.RawMessage) error) (func(), error) + }) + if !ok { + err = errors.New("this tool runner's bus makes no subscription for new messages") + break + } + var stop func() + stop, err = newer.SubscribeNew(said.Pattern, func(envelope json.RawMessage) error { + _, err := c.ask(module, "mesh/event", map[string]any{"envelope": envelope}, EventTimeout) + return err + }) + if err == nil { + c.mu.Lock() + if c.why != nil { + c.mu.Unlock() + stop() + } else { + c.watches = append(c.watches, stop) + c.mu.Unlock() + } + } + break + } mu.Lock() subscriber = c mu.Unlock() diff --git a/node-tools/internal/launch/sessions_test.go b/node-tools/internal/launch/sessions_test.go index e4a1992..f46a5a2 100644 --- a/node-tools/internal/launch/sessions_test.go +++ b/node-tools/internal/launch/sessions_test.go @@ -2,6 +2,7 @@ package launch import ( "encoding/json" + "errors" "os" "path/filepath" "strings" @@ -66,3 +67,77 @@ func TestABundleReadsTheCapabilityWordsRunsAPutSessionAndItsChannelsClosingDrops t.Fatalf("the runner took %v", b.chunks) } } + +// findingBundle asks, as mesh-sdk's finding writer does, a publish with ack that the bus takes, one it refuses, and a +// subscription for new messages; each answer it gets is written to $FINDING_OUT, and it ends after the subscription's. +const findingBundle = `#!/bin/sh +while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p') + case "$line" in + *'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;; + *'"method":"tools/list"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id" + printf '{"jsonrpc":"2.0","id":"a1","method":"mesh/publish","params":{"key":"taken","ack":true,"body":{}}}\n' ;; + *'"id":"a1"'*) printf '%s\n' "$line" >>"$FINDING_OUT" + printf '{"jsonrpc":"2.0","id":"a2","method":"mesh/publish","params":{"key":"refused","ack":true,"body":{}}}\n' ;; + *'"id":"a2"'*) printf '%s\n' "$line" >>"$FINDING_OUT" + printf '{"jsonrpc":"2.0","id":"s1","method":"mesh/subscribe","params":{"pattern":"mesh-conditions.findings-asked","deliver":"new"}}\n' ;; + *'"id":"s1"'*) printf '%s\n' "$line" >>"$FINDING_OUT"; exit 0 ;; + esac +done +` + +type findingBus struct { + subscribing + mu sync.Mutex + pattern string + stopped chan struct{} +} + +func (b *findingBus) PublishAcked(params json.RawMessage) (uint64, error) { + if strings.Contains(string(params), `"refused"`) { + return 0, errors.New("the bus refused it") + } + return 42, nil +} + +func (b *findingBus) SubscribeNew(pattern string, _ func(json.RawMessage) error) (func(), error) { + b.mu.Lock() + b.pattern = pattern + b.mu.Unlock() + return func() { close(b.stopped) }, nil +} + +// Design 47 G §559: `mesh/publish` with `ack` is answered with its sequence once the bus acknowledged it, or `{error}`; +// `mesh/subscribe` with `deliver: new` makes a subscription for new messages that ends with the bundle's channel. +func TestAPublishWithAckIsAnsweredWithItsSequenceAndASubscriptionForNewMessagesEndsWithTheChannel(t *testing.T) { + dir := t.TempDir() + entry, out := filepath.Join(dir, "bundle"), filepath.Join(dir, "out") + if err := os.WriteFile(entry, []byte(findingBundle), 0o755); err != nil { + t.Fatal(err) + } + b := &findingBus{stopped: make(chan struct{})} + l, err := Start("writer", entry, append(os.Environ(), "FINDING_OUT="+out), b, func(string, ...any) {}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(l.Stop) + select { + case <-b.stopped: + case <-time.After(5 * time.Second): + t.Fatal("the subscription for new messages was not ended with the bundle's channel") + } + got, _ := os.ReadFile(out) + for _, want := range []string{`"sequence":42`, `"error":"the bus refused it"`} { + if !strings.Contains(string(got), want) { + t.Errorf("the bundle was answered %s, want it to hold %s", got, want) + } + } + b.mu.Lock() + defer b.mu.Unlock() + if b.pattern != "mesh-conditions.findings-asked" { + t.Errorf("the subscription was made for %q", b.pattern) + } + if len(b.published) != 0 { + t.Errorf("a publish with ack went the unacknowledged way: %v", b.published) + } +} diff --git a/node-tools/internal/runtime/announce_test.go b/node-tools/internal/runtime/announce_test.go index 50fea4a..86356b7 100644 --- a/node-tools/internal/runtime/announce_test.go +++ b/node-tools/internal/runtime/announce_test.go @@ -142,7 +142,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,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" { + if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,finding-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" { 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 d85a250..ccf7ed7 100644 --- a/node-tools/internal/runtime/busstate.go +++ b/node-tools/internal/runtime/busstate.go @@ -40,9 +40,10 @@ var callFamilies = map[string][]string{ } // CallCapabilities is every capability of the calls this runner answers, sorted. The delayed settle of a pulled seat -// message names a fate, not a method, so it stands beside the method families (novox/hq ADR 0664 §5). +// message names a fate, not a method, so it stands beside the method families (novox/hq ADR 0664 §5); so does `finding-calls`, +// the acknowledged `mesh/publish` and the `mesh/subscribe` for new messages the launch answers (design 47 G §559). func CallCapabilities() []string { - out := []string{stdio.CapSeatSettleDelay} + out := []string{stdio.CapSeatSettleDelay, stdio.CapFindingCalls} for c := range callFamilies { out = append(out, c) } diff --git a/node-tools/internal/runtime/ledgers_test.go b/node-tools/internal/runtime/ledgers_test.go index 2fdbf66..324d054 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,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" { + if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,finding-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" { t.Fatalf("the runner states %s", got) } for capability, methods := range callFamilies { diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index bc982f5..c95d14e 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -876,6 +876,21 @@ func (b *moduleBus) Publish(params json.RawMessage) error { return b.all.conn.PublishAs(b.module, env) } +// PublishAcked is Publish that answers the sequence the bus acknowledged the event at (`mesh/publish` with `ack`, design 47 G §559). +func (b *moduleBus) PublishAcked(params json.RawMessage) (uint64, error) { + var env bus.Envelope + if err := json.Unmarshal(params, &env); err != nil { + return 0, fmt.Errorf("not an event envelope: %w", err) + } + return b.all.conn.PublishAsAcked(b.module, env) +} + +// SubscribeNew hands the bundle one event's messages published from now on, on an ordered consumer of its own (`mesh/subscribe` +// with `deliver: new`, design 47 G §559); the consumer ends with the bundle's channel. +func (b *moduleBus) SubscribeNew(pattern string, deliver func(json.RawMessage) error) (func(), error) { + return b.all.conn.SubscribeNew(pattern, deliver) +} + // Ask calls a tool as the module: `{key, body}`, answered with the tool's result (ADR 0198). func (b *moduleBus) Ask(params json.RawMessage) (json.RawMessage, error) { var asked struct { diff --git a/node-tools/internal/runtime/subscribenew_integration_test.go b/node-tools/internal/runtime/subscribenew_integration_test.go new file mode 100644 index 0000000..9dc1d14 --- /dev/null +++ b/node-tools/internal/runtime/subscribenew_integration_test.go @@ -0,0 +1,84 @@ +//go:build integration + +package runtime + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// Design 47 G §559, A-153: against the real bus a publish with ack answers the sequence the bus gave it, and a subscription +// for new messages hears only what is published after it was made, on the subject mesh-sdk's one rule names for the event; +// a wildcard is refused. +func TestAnAckedPublishAnswersItsSequenceAndANewSubscriptionHearsOnlyWhatComesAfter(t *testing.T) { + mt.New(t) + nc, err := nats.Connect(mt.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := jetstream.New(nc) + ctx := context.Background() + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEATEVENTS", Subjects: []string{"mesh.seat.mesh-conditions.event.>"}}); err != nil { + t.Fatal(err) + } + conn := connect(t, "node-tools", "anchor") + publish := func(body string) { + t.Helper() + if _, err := js.Publish(ctx, "mesh.seat.mesh-conditions.event.findings-asked", []byte(body)); err != nil { + t.Fatal(err) + } + } + publish(`{"before":true}`) + + var mu sync.Mutex + var heard []string + stop, err := conn.SubscribeNew("mesh-conditions.findings-asked", func(raw json.RawMessage) error { + var env bus.Envelope + _ = json.Unmarshal(raw, &env) + mu.Lock() + heard = append(heard, env.Key+" "+string(env.Body)) + mu.Unlock() + return nil + }) + if err != nil { + t.Fatal(err) + } + defer stop() + publish(`{"after":true}`) + publish(`{"after":2}`) + mt.Until(t, func() error { + mu.Lock() + defer mu.Unlock() + if len(heard) < 2 { + return context.DeadlineExceeded + } + return nil + }) + mu.Lock() + if len(heard) != 2 || heard[0] != `mesh-conditions.findings-asked {"after":true}` { + t.Errorf("heard %v, want the messages after the subscription alone, in order", heard) + } + mu.Unlock() + + for name, want := range map[string]string{"mesh-conditions.*": "[A-Za-z0-9_-]"} { + if _, err := conn.SubscribeNew(name, func(json.RawMessage) error { return nil }); err == nil || !strings.Contains(err.Error(), want) { + t.Errorf("a subscription for %s answered %v, want one naming %q", name, err, want) + } + } + + first, err := conn.PublishAsAcked("node-tools", bus.Envelope{Key: "finding-written", Body: json.RawMessage(`{}`)}) + second, err2 := conn.PublishAsAcked("node-tools", bus.Envelope{Key: "finding-written", Body: json.RawMessage(`{}`)}) + if err != nil || err2 != nil || first == 0 || second != first+1 { + t.Errorf("acknowledged publishes answered %d (%v) and %d (%v), want consecutive sequences", first, err, second, err2) + } +} diff --git a/node-tools/module.json b/node-tools/module.json index 076c37f..fbc4950 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -65,6 +65,7 @@ "binary": "node-tools", "states": [ "bus-state-calls", + "finding-calls", "ledger-calls", "ledger-upkeep-calls", "put-sessions",