S15: sm-leaf health HTTP, verae.sm.* acks, systemd unit
Some checks are pending
ci / python (push) Waiting to run
ci / go (push) Waiting to run

Loopback GET /health, nats-leaf:// rewrite, subject request-reply tests.
Do not replace pfc-py-admin. NpeRequired stays on the package surface.
This commit is contained in:
George Lambert 2026-09-15 22:40:50 -04:00
parent 9eed0b1942
commit f4da7ff446
7 changed files with 158 additions and 11 deletions

View file

@ -2,8 +2,12 @@
package leaf
import (
"encoding/json"
"errors"
"log"
"net/http"
"net/url"
"strings"
"time"
natsserver "github.com/nats-io/nats-server/v2/server"
@ -16,11 +20,17 @@ type Node struct {
nc *nats.Conn
}
// ParseHub rewrites nats-leaf:// to nats:// then parses the URL.
func ParseHub(hub string) (*url.URL, error) {
raw := strings.Replace(hub, "nats-leaf://", "nats://", 1)
return url.Parse(raw)
}
// Start binds 127.0.0.1:0 and optionally leaf-connects to hub.
func Start(hub string) (*Node, error) {
opts := &natsserver.Options{Host: "127.0.0.1", Port: -1}
if hub != "" {
u, err := url.Parse(hub)
u, err := ParseHub(hub)
if err != nil {
return nil, err
}
@ -40,17 +50,61 @@ func Start(hub string) (*Node, error) {
return nil, err
}
n := &Node{ns: ns, nc: nc}
if _, err := n.nc.Subscribe("verae.sm.send", func(msg *nats.Msg) {
if msg.Reply != "" {
_ = n.nc.Publish(msg.Reply, []byte(`{"accepted":true}`))
}
}); err != nil {
if err := n.subscribe(); err != nil {
n.Shutdown()
return nil, err
}
return n, nil
}
func (n *Node) subscribe() error {
ack := func(msg *nats.Msg, body []byte) {
if msg.Reply != "" {
_ = n.nc.Publish(msg.Reply, body)
}
}
if _, err := n.nc.Subscribe("verae.sm.send", func(msg *nats.Msg) {
// Passthrough: do not log ciphertext.
ack(msg, []byte(`{"accepted":true}`))
}); err != nil {
return err
}
if _, err := n.nc.Subscribe("verae.sm.dead", func(msg *nats.Msg) {
ack(msg, []byte(`{"queued":true}`))
}); err != nil {
return err
}
if _, err := n.nc.Subscribe("verae.sm.error", func(msg *nats.Msg) {
ack(msg, []byte(`{"emitted":true}`))
}); err != nil {
return err
}
if _, err := n.nc.Subscribe("verae.sm.log.summary", func(msg *nats.Msg) {
var hdr map[string]any
_ = json.Unmarshal(msg.Data, &hdr)
log.Printf("sm-summary code=%v lookup=%v class=%v", hdr["error_code"], hdr["lookup_id"], hdr["dest_class"])
ack(msg, []byte(`{"logged":true}`))
}); err != nil {
return err
}
return nil
}
// HealthHandler is the loopback JSON health mux (does not expose NATS).
func HealthHandler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"ok":true,"service":"sm-leaf"}`))
})
return mux
}
// ServeHealth binds a loopback health listener (does not expose NATS).
func ServeHealth(addr string) error {
return http.ListenAndServe(addr, HealthHandler())
}
// Addr is the in-process client URL.
func (n *Node) Addr() string { return n.ns.ClientURL() }

View file

@ -1,6 +1,12 @@
package leaf
import "testing"
import (
"encoding/json"
"io"
"net/http/httptest"
"testing"
"time"
)
func TestStartInProcess(t *testing.T) {
n, err := Start("")
@ -12,3 +18,57 @@ func TestStartInProcess(t *testing.T) {
t.Fatal("empty addr")
}
}
func TestParseHubLeafScheme(t *testing.T) {
u, err := ParseHub("nats-leaf://10.10.10.21:7422")
if err != nil {
t.Fatal(err)
}
if u.Scheme != "nats" {
t.Fatalf("scheme %s", u.Scheme)
}
if u.Host != "10.10.10.21:7422" {
t.Fatalf("host %s", u.Host)
}
}
func TestHealthHandler(t *testing.T) {
srv := httptest.NewServer(HealthHandler())
defer srv.Close()
resp, err := srv.Client().Get(srv.URL + "/health")
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
var m map[string]any
if err := json.Unmarshal(body, &m); err != nil {
t.Fatal(err)
}
if m["ok"] != true || m["service"] != "sm-leaf" {
t.Fatalf("health %s", body)
}
}
func TestSMSubjectsAck(t *testing.T) {
n, err := Start("")
if err != nil {
t.Fatal(err)
}
defer n.Shutdown()
cases := map[string]string{
"verae.sm.send": `{"accepted":true}`,
"verae.sm.dead": `{"queued":true}`,
"verae.sm.error": `{"emitted":true}`,
"verae.sm.log.summary": `{"logged":true}`,
}
for subj, want := range cases {
msg, err := n.nc.Request(subj, []byte(`{"lookup_id":"t"}`), time.Second)
if err != nil {
t.Fatalf("%s: %v", subj, err)
}
if string(msg.Data) != want {
t.Fatalf("%s got %s want %s", subj, msg.Data, want)
}
}
}