jschoubben
49b307a46d
The lease write asked seat:node-service-manager.stop@novox-be through Conn.Call, which published the '@' inside the subject: nobody served it and the call timed out after 30 s with nothing said (issue 774 notes 84-87). Every key now parses in one place, calladdress: Call routes '@node' as CallOn does, and stdio Ask refuses a key the runtime could not route, before anything is sent. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
284 lines
8.9 KiB
Go
284 lines
8.9 KiB
Go
//go:build integration
|
|
|
|
package process
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
)
|
|
|
|
// 1191/AC1: given a stand-in holder answering seat operator-channel verb tell on a test bus, when the
|
|
// process connection of novox.mesh-mail.send calls seat:operator-channel.tell with {headline, lines,
|
|
// subject}, then the request goes to mesh.seat.operator-channel.call.tell.novox~mesh-mail~send with the
|
|
// arguments as the body, and the holder's {result} comes back; a reply with error set comes back as an
|
|
// error with those words.
|
|
func TestCallReachesTheCallerNamedSubjectAndReadsTheReply(t *testing.T) {
|
|
url, account := grantedSendProcess(t)
|
|
adminNC, _ := adminJS(t, url)
|
|
|
|
heard := make(chan *nats.Msg, 1)
|
|
sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
|
|
heard <- m
|
|
_ = m.Respond([]byte(`{"result":{"sent":true}}`))
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _ = sub.Unsubscribe() }()
|
|
if err := adminNC.Flush(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
conn, err := Connect(account)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
|
|
result, err := conn.Call("seat:operator-channel.tell",
|
|
map[string]any{"headline": "h", "lines": []string{"l1"}, "subject": "s"})
|
|
if err != nil {
|
|
t.Fatalf("Call: %v", err)
|
|
}
|
|
if string(result) != `{"sent":true}` {
|
|
t.Fatalf("result: %s", result)
|
|
}
|
|
|
|
select {
|
|
case msg := <-heard:
|
|
var body map[string]any
|
|
if err := json.Unmarshal(msg.Data, &body); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if body["headline"] != "h" || body["subject"] != "s" {
|
|
t.Errorf("the arguments were not the call's body: %v", body)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("the holder never heard the call")
|
|
}
|
|
}
|
|
|
|
// 1191/AC1: a reply with error set comes back as an error with those words.
|
|
func TestCallReadsAnErrorReplyAsAnError(t *testing.T) {
|
|
url, account := grantedSendProcess(t)
|
|
adminNC, _ := adminJS(t, url)
|
|
sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
|
|
_ = m.Respond([]byte(`{"error":"refused: no such warrant"}`))
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _ = sub.Unsubscribe() }()
|
|
if err := adminNC.Flush(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
conn, err := Connect(account)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
|
|
_, err = conn.Call("seat:operator-channel.tell", map[string]any{})
|
|
if err == nil || err.Error() != "refused: no such warrant" {
|
|
t.Fatalf("Call with an error reply: %v, want the words of the error", err)
|
|
}
|
|
}
|
|
|
|
// 1191/AC2: given no responder on the caller-named subject, when the call is made, then it is asked
|
|
// again on mesh.seat.operator-channel.tool.tell (the tool runner's fallback).
|
|
func TestCallFallsBackToThePlainSubjectWhenNothingAnswersTheCallerNamedOne(t *testing.T) {
|
|
url, account := grantedSendProcess(t)
|
|
adminNC, _ := adminJS(t, url)
|
|
sub, err := adminNC.Subscribe("mesh.seat.operator-channel.tool.tell", func(m *nats.Msg) {
|
|
_ = m.Respond([]byte(`{"result":{"via":"fallback"}}`))
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _ = sub.Unsubscribe() }()
|
|
if err := adminNC.Flush(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
conn, err := Connect(account)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
|
|
result, err := conn.Call("seat:operator-channel.tell", map[string]any{})
|
|
if err != nil {
|
|
t.Fatalf("Call: %v", err)
|
|
}
|
|
if string(result) != `{"via":"fallback"}` {
|
|
t.Fatalf("result: %s, want the fallback holder's answer", result)
|
|
}
|
|
}
|
|
|
|
// 1191/AC4, release 2's acceptance test: given the test bus with accounts composed as the controller
|
|
// composes a process account (the subjects of mesh-controller internal/broker/process_grants.go), when
|
|
// one process connection emits, updates a state, reads an own ask's record, consumes and acknowledges its
|
|
// queue and calls tell, then each succeeds; and its attempt to subscribe a tool-call subject is refused
|
|
// by the bus.
|
|
func TestReleaseAcceptanceAProcessDoesExactlyItsBusDeclaration(t *testing.T) {
|
|
url, account := grantedSendProcess(t)
|
|
_, adminJetStream := adminJS(t, url)
|
|
|
|
if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil {
|
|
t.Fatalf("the admin could not make the state bucket: %v", err)
|
|
}
|
|
asksKV, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "messenger_asks"})
|
|
if err != nil {
|
|
t.Fatalf("the admin could not make the records bucket: %v", err)
|
|
}
|
|
if _, err := asksKV.Put(ctxFor(t), "mesh-mail.ask-1", []byte(`{"state":"decided"}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{
|
|
Name: "MAIL_EVENTS", Subjects: []string{"mesh.mod.mesh-mail.event.>"}, Duplicates: time.Minute,
|
|
}); err != nil {
|
|
t.Fatalf("the admin could not make the events stream: %v", err)
|
|
}
|
|
makeQueue(t, url, account, 30*time.Second)
|
|
publishApproved(t, url, `{"send":"s1"}`)
|
|
|
|
adminNC, _ := adminJS(t, url)
|
|
holderSub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
|
|
_ = m.Respond([]byte(`{"result":{}}`))
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _ = holderSub.Unsubscribe() }()
|
|
if err := adminNC.Flush(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
conn, err := Connect(account)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
|
|
// 1. Emits.
|
|
if err := conn.Emit("mail.send.sent", map[string]any{"send": "s1"}, stdio.EmitOptions{MessageID: "ev-1"}); err != nil {
|
|
t.Fatalf("Emit: %v", err)
|
|
}
|
|
|
|
// 2. Updates a state.
|
|
s := conn.State("sends")
|
|
rev, err := s.Create("k", map[string]any{"v": 1})
|
|
if err != nil {
|
|
t.Fatalf("State.Create: %v", err)
|
|
}
|
|
if _, err := s.Update("k", map[string]any{"v": 2}, rev); err != nil {
|
|
t.Fatalf("State.Update: %v", err)
|
|
}
|
|
|
|
// 3. Reads an own ask's record.
|
|
if e, err := conn.Ask("messenger_asks", "ask-1"); err != nil || e == nil {
|
|
t.Fatalf("Ask: %+v, %v", e, err)
|
|
}
|
|
|
|
// 4. Consumes and acknowledges its queue.
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
consumed := make(chan error, 1)
|
|
go func() {
|
|
consumed <- conn.Consume(ctx, "MAIL_SENDS", func(m *Message) {
|
|
if err := m.Ack(); err != nil {
|
|
t.Errorf("Ack: %v", err)
|
|
}
|
|
cancel()
|
|
})
|
|
}()
|
|
select {
|
|
case <-consumed:
|
|
case <-time.After(8 * time.Second):
|
|
t.Fatal("the process did not consume its own queue")
|
|
}
|
|
|
|
// 5. Calls tell.
|
|
if _, err := conn.Call("seat:operator-channel.tell", map[string]any{}); err != nil {
|
|
t.Fatalf("Call: %v", err)
|
|
}
|
|
|
|
// 6. Refused: subscribing a tool-call subject — a fresh connection, so the error handler catches
|
|
// exactly this one refusal and nothing earlier.
|
|
refusals := make(chan error, 1)
|
|
watched, err := Connect(account, nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err }))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer watched.Close()
|
|
if _, err := watched.nc.Subscribe("mesh.mod.mesh-mail.tool.>", func(*nats.Msg) {}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := watched.nc.Flush(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
select {
|
|
case err := <-refusals:
|
|
if !errors.Is(err, nats.ErrPermissionViolation) {
|
|
t.Errorf("subscribing a tool-call subject was not refused as a permissions violation: %v", err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Error("subscribing a tool-call subject the process does not serve was not refused")
|
|
}
|
|
}
|
|
|
|
// 1470/AC4: two answerers of the same verb on the subjects of nodes ace and g14; CallOn naming g14 gets g14's
|
|
// answer, and ace's answerer hears nothing. Issue 774 notes 84-87: Call with the key `<verb>@g14` is published on the
|
|
// same subject, never on one holding the '@' that nobody serves.
|
|
func TestCallOnReachesTheNamedNodeAlone(t *testing.T) {
|
|
url, account := grantedSendProcess(t)
|
|
adminNC, _ := adminJS(t, url)
|
|
aceHeard := make(chan struct{}, 1)
|
|
for _, node := range []string{"ace", "g14"} {
|
|
node := node
|
|
sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell."+node+".novox~mesh-mail~send", func(m *nats.Msg) {
|
|
if node == "ace" {
|
|
aceHeard <- struct{}{}
|
|
}
|
|
_ = m.Respond([]byte(`{"result":{"node":"` + node + `"}}`))
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _ = sub.Unsubscribe() }()
|
|
}
|
|
if err := adminNC.Flush(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
conn, err := Connect(account)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
|
|
result, err := conn.CallOn("g14", "seat:operator-channel.tell", map[string]any{})
|
|
if err != nil {
|
|
t.Fatalf("CallOn: %v", err)
|
|
}
|
|
if string(result) != `{"node":"g14"}` {
|
|
t.Fatalf("result: %s, want g14's answer", result)
|
|
}
|
|
if result, err = conn.Call("seat:operator-channel.tell@g14", map[string]any{}); err != nil || string(result) != `{"node":"g14"}` {
|
|
t.Fatalf("Call with @g14: %s, %v, want g14's answer", result, err)
|
|
}
|
|
select {
|
|
case <-aceHeard:
|
|
t.Error("ace's answerer heard a call named for g14")
|
|
default:
|
|
}
|
|
}
|