From 94702bcd79197a18e8a54552d287beb090adccff Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 04:35:33 +0200 Subject: [PATCH] Let a process read the router's record of its module's own ask (issue 1189) --- go/process/asks.go | 68 +++++++++++++++++++++++++++++ go/process/asks_integration_test.go | 60 +++++++++++++++++++++++++ go/process/asks_test.go | 28 ++++++++++++ 3 files changed, 156 insertions(+) create mode 100644 go/process/asks.go create mode 100644 go/process/asks_integration_test.go create mode 100644 go/process/asks_test.go diff --git a/go/process/asks.go b/go/process/asks.go new file mode 100644 index 0000000..0aae29b --- /dev/null +++ b/go/process/asks.go @@ -0,0 +1,68 @@ +package process + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// Ask reads the router's record of one of this process's module's own asks, under its own module and no +// other's (novox/hq ADR 0436 clause 4, ADR 0259 §3): by direct get on bucket, the same subject the tool +// runner reads (mesh-tools node-tools/internal/bus/seat.go ~300-345): +// +// $JS.API.DIRECT.GET.KV_.$KV... +// +// bucket is the router's records bucket as the caller's own manifest names it (operator-channel's is +// messenger_asks in the controller's own test fixtures) — never hard-coded to one bucket here, since a +// process may read records of more than one seat's router. A missing, deleted or purged key is nil, not +// an error. +func (c *Conn) Ask(bucket, id string) (*stdio.StateEntry, error) { + if err := checkAskID(id); err != nil { + return nil, err + } + subject := "$JS.API.DIRECT.GET.KV_" + bucket + ".$KV." + bucket + "." + c.account.Module + "." + id + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + reply, err := c.nc.RequestWithContext(ctx, subject, nil) + if err != nil { + return nil, fmt.Errorf("reading %s's ask %s from %s: %w", c.account.Module, id, bucket, err) + } + if status := reply.Header.Get("Status"); status == "404" { + return nil, nil + } else if status != "" { + return nil, fmt.Errorf("the record %s.%s could not be read: %s %s", c.account.Module, id, status, + reply.Header.Get("Description")) + } + if op := reply.Header.Get("KV-Operation"); op == "DEL" || op == "PURGE" { + return nil, nil + } + var revision uint64 + if _, err := fmt.Sscan(reply.Header.Get("Nats-Sequence"), &revision); err != nil { + return nil, fmt.Errorf("the record %s.%s carries no readable revision: %w", c.account.Module, id, err) + } + return &stdio.StateEntry{Key: c.account.Module + "." + id, Value: askValueOf(reply.Data), Revision: revision}, nil +} + +// checkAskID refuses an id holding a dot or a wildcard before anything is sent: either would name a +// different key under the direct-get subject's own dot-joined path, never this one (novox/hq issue 1189 +// AC2). +func checkAskID(id string) error { + if id == "" || strings.ContainsAny(id, ".*>") { + return fmt.Errorf("%q is not an ask id: an ask id holds no dot and no wildcard, naming one key and "+ + "no other", id) + } + return nil +} + +// askValueOf is a record's raw bytes as JSON, quoting them when they are not already valid JSON +// (mesh-tools node-tools/internal/bus/state.go:351-357, valueOf). +func askValueOf(raw []byte) json.RawMessage { + if len(raw) == 0 || !json.Valid(raw) { + b, _ := json.Marshal(string(raw)) + return b + } + return json.RawMessage(append([]byte(nil), raw...)) +} diff --git a/go/process/asks_integration_test.go b/go/process/asks_integration_test.go new file mode 100644 index 0000000..de84be7 --- /dev/null +++ b/go/process/asks_integration_test.go @@ -0,0 +1,60 @@ +//go:build integration + +package process + +import ( + "testing" + + "github.com/nats-io/nats.go/jetstream" +) + +// 1189/AC1: given the router's records bucket on a test bus holding a record under ., +// when the process connection reads its module's ask of that id, then it gets the record's bytes and +// revision by direct get; a deleted or purged key is said as absent; a missing key as absent. +func TestAskReadsTheRoutersRecordOfItsOwnAsk(t *testing.T) { + url, account := grantedSendProcess(t) + _, adminJetStream := adminJS(t, url) + kv, 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) + } + rev, err := kv.Put(ctxFor(t), "mesh-mail.ask-1", []byte(`{"state":"decided"}`)) + if err != nil { + t.Fatal(err) + } + + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + e, err := conn.Ask("messenger_asks", "ask-1") + if err != nil { + t.Fatalf("reading an existing record: %v", err) + } + if e == nil || e.Revision != rev || string(e.Value) != `{"state":"decided"}` { + t.Fatalf("read back %+v, want revision %d and {\"state\":\"decided\"}", e, rev) + } + + if err := kv.Delete(ctxFor(t), "mesh-mail.ask-1"); err != nil { + t.Fatal(err) + } + if e, err := conn.Ask("messenger_asks", "ask-1"); err != nil || e != nil { + t.Fatalf("a deleted record: %+v, %v, want absent", e, err) + } + + if _, err := kv.Create(ctxFor(t), "mesh-mail.ask-2", []byte(`{"state":"asked"}`)); err != nil { + t.Fatal(err) + } + if err := kv.Purge(ctxFor(t), "mesh-mail.ask-2"); err != nil { + t.Fatal(err) + } + if e, err := conn.Ask("messenger_asks", "ask-2"); err != nil || e != nil { + t.Fatalf("a purged record: %+v, %v, want absent", e, err) + } + + if e, err := conn.Ask("messenger_asks", "ask-missing"); err != nil || e != nil { + t.Fatalf("a missing record: %+v, %v, want absent", e, err) + } +} diff --git a/go/process/asks_test.go b/go/process/asks_test.go new file mode 100644 index 0000000..6e564cb --- /dev/null +++ b/go/process/asks_test.go @@ -0,0 +1,28 @@ +package process + +import "testing" + +// 1189/AC2: given an ask id holding a dot or a wildcard, when read, then it is refused before anything is +// sent — proven on a Conn with no live connection at all: if Ask tried to send, it would panic on the nil +// *nats.Conn instead of returning a clean error. +func TestAskRefusesAnInvalidIDBeforeSendingAnything(t *testing.T) { + c := &Conn{account: Account{Module: "mesh-mail"}} + for _, id := range []string{"", "a.b", "a*", "a>", "ask.1", "ask.*"} { + if _, err := c.Ask("messenger_asks", id); err == nil { + t.Errorf("Ask(%q) was not refused", id) + } + } +} + +func TestCheckAskID(t *testing.T) { + for _, id := range []string{"", "a.b", "a*b", "a>b", ".", "*"} { + if err := checkAskID(id); err == nil { + t.Errorf("checkAskID(%q) was accepted", id) + } + } + for _, id := range []string{"ask-1", "a1b2c3", "send_42"} { + if err := checkAskID(id); err != nil { + t.Errorf("checkAskID(%q) was refused: %v", id, err) + } + } +}