A3f item 2: one whole test of a process holder's marked verb on the bus

Against an embedded bus, a module's process calls a holder's marked verb through process.Conn's call: asked, approved, run once, recorded through mesh-hand-acts.warranted and its answer {answer, approval} heard on its own gated-ended; a declined call runs nothing; a marked verb served through no gate, or a gate whose state approvals is not on the bus, fails closed. Builds: design 47 G §677, §700, lead answers A-158, A-224, A-244 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
2026-11-12 09:31:08 +01:00
parent d3980ded07
commit 89359111a9
1 changed files with 261 additions and 0 deletions
@@ -0,0 +1,261 @@
+//go:build integration
+
+package process
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/approval"
+	"git.novox.be/novox/mesh-sdk/go/operatorask"
+	"git.novox.be/novox/mesh-sdk/go/seatdef"
+)
+
+// busChannel is the operator channel as the test's router: the gate's asks kept in the holder's state approvals, the
+// router's record read under the holder's name from messenger_asks (design 47 G §700), the claim as
+// operatorchannelclient's Take makes it. asked says each ask raised.
+type busChannel struct {
+	conn  *Conn
+	state State
+	asked chan approval.Ask
+}
+
+func (b *busChannel) Ask(module string, a approval.Ask) error {
+	raw, _ := json.Marshal(a)
+	if _, err := b.state.Create("ask."+a.ID, approval.Kept[json.RawMessage, json.RawMessage]{Ask: raw, Asked: time.Now()}); err != nil {
+		return err
+	}
+	b.asked <- a
+	return nil
+}
+
+func (b *busChannel) Record(module, id string) (*approval.Record, error) {
+	e, err := b.conn.Ask("messenger_asks", id)
+	if err != nil || e == nil {
+		return nil, err
+	}
+	var r approval.Record
+	return &r, json.Unmarshal(e.Value, &r)
+}
+
+func (b *busChannel) Claim(module string, w approval.Warrant, act operatorask.Act, run func()) (approval.Claim, error) {
+	e, err := b.state.Get("ask." + w.Ask)
+	if err != nil || e == nil {
+		return approval.ClaimRefused, fmt.Errorf("no ask %s: %v", w.Ask, err)
+	}
+	var k approval.Kept[json.RawMessage, json.RawMessage]
+	_ = json.Unmarshal(e.Value, &k)
+	if k.Acted {
+		return approval.ClaimDone, nil
+	}
+	var a approval.Ask
+	_ = json.Unmarshal(k.Ask, &a)
+	if d, _ := operatorask.ActDigest(act); d != a.About {
+		return approval.ClaimRefused, fmt.Errorf("the option binds another act")
+	}
+	k.Acted = true
+	if _, err := b.state.Update("ask."+w.Ask, k, e.Revision); err != nil {
+		return approval.ClaimDone, nil
+	}
+	run()
+	return approval.Claimed, nil
+}
+
+func (b *busChannel) Cancel(string, string) error { return nil }
+func (b *busChannel) Decided(module string) string {
+	return "mesh.seat.operator-channel.event.decided." + module
+}
+func (b *busChannel) Seat() string { return "operator-channel" }
+
+// Design 47 G §677 (lead answers A-158, A-224, A-244), its one whole test: a process holder's marked verb, called by a
+// module's process through process.Conn's call, is asked, approved, run once, recorded in the hand-act log and its
+// answer heard on the caller's own gated-ended; a declined call runs nothing; a marked verb served through no gate, or
+// a gate whose state approvals is not there, fails closed.
+func TestAProcessHoldersMarkedVerbIsAskedApprovedRunAndRecorded(t *testing.T) {
+	opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true,
+		Users: []*server.User{{Username: adminUser, Password: adminPass}, {Username: "n1.holder", Password: "pw"},
+			{Username: "n1.caller", Password: "pw"}, {Username: "n1.mesh-hand-acts", Password: "pw"}, {Username: "n1.other", Password: "pw"}}}
+	s, err := server.NewServer(opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	go s.Start()
+	if !s.ReadyForConnections(10 * time.Second) {
+		t.Fatal("the test bus did not come up")
+	}
+	t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
+	url := s.ClientURL()
+	_, js := adminJS(t, url)
+	ctx := ctxFor(t)
+	records, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "messenger_asks"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "holder_" + approval.StateName, TTL: 48 * time.Hour}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: EventsStream, Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.operator-channel.event.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.CreateOrUpdateConsumer(ctx, EventsStream, jetstream.ConsumerConfig{Durable: "n1_holder", AckPolicy: jetstream.AckExplicitPolicy,
+		FilterSubjects: []string{"mesh.seat.operator-channel.event.decided.holder"}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: seatdef.KeptStream("wiper", seatdef.Message{Name: approval.GatedEndedEvent}),
+		Subjects: []string{"mesh.seat.wiper.event.gated-ended.>"}, MaxMsgsPerSubject: 1, MaxAge: approval.GatedEndedKeptFor}); err != nil {
+		t.Fatal(err)
+	}
+	connect := func(module string) *Conn {
+		c, err := Connect(Account{URL: url, Node: "n1", Module: module, User: "n1." + module, Password: "pw"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(c.Close)
+		return c
+	}
+
+	// The seat mesh-hand-acts' holder, recording each warranted run.
+	var mu sync.Mutex
+	var warranted []string
+	handActs := connect("mesh-hand-acts")
+	if _, err := handActs.AnswerSeat("mesh-hand-acts", func() (seatdef.Definition, bool) {
+		return seatdef.Definition{Seat: "mesh-hand-acts", Verbs: []seatdef.Verb{{Name: "warranted"}}}, true
+	}, map[string]SeatHandler{"warranted": func(caller string, body json.RawMessage) (any, error) {
+		mu.Lock()
+		defer mu.Unlock()
+		warranted = append(warranted, caller+" "+string(body))
+		return map[string]bool{"recorded": true}, nil
+	}}); err != nil {
+		t.Fatal(err)
+	}
+
+	// The holder: its marked verb wipe behind its gate, its decided heard through its hub, its start read.
+	holder := connect("holder")
+	ch := &busChannel{conn: holder, state: holder.State(approval.StateName), asked: make(chan approval.Ask, 4)}
+	gate := holder.Gate(Gating{Channel: ch, Build: "fp1", Logf: t.Logf})
+	ran := 0
+	marked := func(seat string) func() (seatdef.Definition, bool) {
+		return func() (seatdef.Definition, bool) {
+			return seatdef.Definition{Seat: seat, Verbs: []seatdef.Verb{{Name: "wipe", NeedsApproval: true}, {Name: "approvals"}}}, true
+		}
+	}
+	wipe := map[string]SeatHandler{"wipe": func(caller string, body json.RawMessage) (any, error) {
+		mu.Lock()
+		defer mu.Unlock()
+		ran++
+		return "wiped", nil
+	}}
+	if _, err := holder.AnswerSeat("wiper", marked("wiper"), wipe, Gated(gate)); err != nil {
+		t.Fatal(err)
+	}
+	hub := NewEventHub(t.Logf)
+	if err := HearDecided(gate, ch, "holder", hub); err != nil {
+		t.Fatal(err)
+	}
+	hjs, _ := jetstream.New(holder.nc)
+	if err := hub.Start(ctx, hjs, holder.account); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(hub.Stop)
+	if err := gate.StartHolding(); err != nil {
+		t.Fatal(err)
+	}
+
+	caller := connect("caller")
+	type result struct {
+		answer json.RawMessage
+		err    error
+	}
+	callWipe := func(args map[string]string) chan result {
+		done := make(chan result, 1)
+		go func() {
+			a, err := caller.Call("seat:wiper.wipe", args)
+			done <- result{a, err}
+		}()
+		return done
+	}
+	decide := func(a approval.Ask, option string) {
+		raw, _ := json.Marshal(approval.Record{State: "chosen", OfferedOn: []string{"telegram"},
+			Warrant: &approval.Warrant{Ask: a.ID, Asker: "holder", Option: option, Channel: "telegram", At: time.Now().UTC().Format(time.RFC3339Nano)}})
+		if _, err := records.Put(ctx, "holder."+a.ID, raw); err != nil {
+			t.Fatal(err)
+		}
+		// The decided's body is not read beyond the ask's id: the gate reads the router's record.
+		if _, err := js.Publish(ctx, ch.Decided("holder"), []byte(`{"ask":"`+a.ID+`","option":"approve"}`)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	wait := func(done chan result) result {
+		select {
+		case r := <-done:
+			return r
+		case <-time.After(10 * time.Second):
+			t.Fatal("the call's end was not heard within 10s")
+			return result{}
+		}
+	}
+	asked := func() approval.Ask {
+		select {
+		case a := <-ch.asked:
+			return a
+		case <-time.After(10 * time.Second):
+			t.Fatal("no ask raised within 10s")
+			return approval.Ask{}
+		}
+	}
+
+	t.Run("asked, approved, run once, recorded, heard on the caller's gated-ended", func(t *testing.T) {
+		done := callWipe(map[string]string{"dir": "scratch"})
+		a := asked()
+		decide(a, approval.OptionApprove)
+		r := wait(done)
+		var got approval.Answered
+		if r.err != nil || json.Unmarshal(r.answer, &got) != nil || got.Answer != "wiped" || !strings.Contains(got.Approval, a.ID) {
+			t.Fatalf("answer %s, %v", r.answer, r.err)
+		}
+		mu.Lock()
+		defer mu.Unlock()
+		if ran != 1 || len(warranted) != 1 || !strings.Contains(warranted[0], a.ID) || !strings.HasPrefix(warranted[0], "n1.holder ") {
+			t.Errorf("ran %d, warranted %v", ran, warranted)
+		}
+	})
+	t.Run("declined, nothing runs", func(t *testing.T) {
+		done := callWipe(map[string]string{"dir": "other"})
+		decide(asked(), approval.OptionDecline)
+		if r := wait(done); r.err == nil || !strings.Contains(r.err.Error(), "declined") {
+			t.Fatalf("answer %s, %v", r.answer, r.err)
+		}
+		mu.Lock()
+		defer mu.Unlock()
+		if ran != 1 {
+			t.Errorf("ran %d", ran)
+		}
+	})
+	t.Run("failing closed: no gate, or no state approvals", func(t *testing.T) {
+		if _, err := holder.AnswerSeat("bare", marked("bare"), wipe); err != nil {
+			t.Fatal(err)
+		}
+		other := connect("other") // its state approvals is not on the bus
+		g := other.Gate(Gating{Channel: &busChannel{conn: other, state: other.State(approval.StateName), asked: make(chan approval.Ask, 1)}, Build: "fp1"})
+		if _, err := other.AnswerSeat("keeper", marked("keeper"), wipe, Gated(g)); err != nil {
+			t.Fatal(err)
+		}
+		for _, seat := range []string{"bare", "keeper"} {
+			if _, err := caller.Call("seat:"+seat+".wipe", map[string]string{"dir": "x"}); err == nil || !strings.Contains(err.Error(), "Not run") {
+				t.Errorf("%s: %v", seat, err)
+			}
+		}
+		mu.Lock()
+		defer mu.Unlock()
+		if ran != 1 {
+			t.Errorf("ran %d", ran)
+		}
+	})
+}