From 89359111a93248dc6b2c1da349ad0009b83ccbb8 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:31:08 +0100 Subject: [PATCH] A3f item 2: one whole test of a process holder's marked verb on the bus MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- go/process/gated_integration_test.go | 261 +++++++++++++++++++++++++++ 1 file changed, 261 insertions(+) create mode 100644 go/process/gated_integration_test.go diff --git a/go/process/gated_integration_test.go b/go/process/gated_integration_test.go new file mode 100644 index 00000000..5ddad2dc --- /dev/null +++ b/go/process/gated_integration_test.go @@ -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) + } + }) +}