From 5ba1da5b821cb60d5280c2d18c76d4b88fc1d1f8 Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 00:28:36 +0000 Subject: [PATCH] Carry the gate's merge asks to the router and write the operator's answer back, a warrant that does not hold said as a condition (issue 1343) --- .../cmd/mesh-delivery-teller/decided.go | 115 ++++++ .../mesh-delivery-teller/integration_test.go | 275 +++++++++++++ .../cmd/mesh-delivery-teller/kv.go | 101 +++++ .../cmd/mesh-delivery-teller/main.go | 54 ++- .../cmd/mesh-delivery-teller/teller.go | 383 ++++++++++++++++++ .../cmd/mesh-delivery-teller/teller_test.go | 375 +++++++++++++++++ modules/mesh-delivery-teller/go.mod | 1 + modules/mesh-delivery-teller/go.sum | 2 + modules/mesh-delivery-teller/manifest_test.go | 16 +- modules/mesh-delivery-teller/module.json | 3 +- 10 files changed, 1314 insertions(+), 11 deletions(-) create mode 100644 modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go create mode 100644 modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go create mode 100644 modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go create mode 100644 modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go new file mode 100644 index 0000000..2d20503 --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go @@ -0,0 +1,115 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// The teller's three bus adapters for mesh-sdk go/asker, which in a bundle reaches the bus through the runtime: here the +// teller is a process on its own credential. What each may do is the controller's composed grant for a module that uses +// the seat operator-channel (mesh-control internal/broker seattraffic.go SeatTrafficOf): publish its own ask, cancel and +// condition, hear its own decided event, and read the router's record of its own asks. + +// callTimeout bounds a read of the router's record. +const callTimeout = 30 * time.Second + +// publishVia publishes an accept of the seat, de-duplicated by id: the same id twice is one message. +func publishVia(ctx context.Context, js jetstream.JetStream) func(subject string, body any, id string) error { + return func(subject string, body any, id string) error { + data, err := json.Marshal(body) + if err != nil { + return err + } + var opts []jetstream.PublishOpt + if id != "" { + opts = append(opts, jetstream.WithMsgID(id)) + } + _, err = js.Publish(ctx, subject, data, opts...) + return err + } +} + +// recordVia reads the router's record of one of this module's asks: the key `.` of the router's bucket asks, +// by a direct get on the subject the controller grants for it, `$JS.API.DIRECT.GET.KV_asks.$KV.asks..>`. The +// record is {state, warrant}. An ask the router holds no record of is nil. +func recordVia(ctx context.Context, conn *nats.Conn, module string) func(id string) (*stdio.StateEntry, error) { + return func(id string) (*stdio.StateEntry, error) { + ctx, cancel := context.WithTimeout(ctx, callTimeout) + defer cancel() + msg, err := conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_asks.$KV.asks."+module+"."+id, nil) + if errors.Is(err, nats.ErrNoResponders) { + return nil, nil + } + if err != nil { + return nil, err + } + if status := msg.Header.Get("Status"); status == "404" || len(msg.Data) == 0 { + return nil, nil + } + return &stdio.StateEntry{Key: id, Value: msg.Data}, nil + } +} + +// decidedVia hears the warrants the router says to this module alone: its own `decided` event, a subject only the router +// publishes (the grant to subscribe it names the module). The handler is the asker's: pattern is the key it expects. +func decidedVia(conn *nats.Conn, module string, logf func(string, ...any)) func(pattern string, handle func(stdio.Envelope) error) error { + return func(pattern string, handle func(stdio.Envelope) error) error { + _, err := conn.Subscribe(asks.DecidedSubject(module), func(m *nats.Msg) { + if err := handle(stdio.Envelope{Key: pattern, Body: m.Data}); err != nil { + logf("a warrant could not be handled and is read again from the router's record at the next start: %v", err) + } + }) + return err + } +} + +// sayVia publishes a condition on the seat's one subject for conditions, under the module's own name. +func sayVia(ctx context.Context, js jetstream.JetStream, module string) func(c asks.Condition) error { + return func(c asks.Condition) error { + subject, err := asks.ConditionSubject(module) + if err != nil { + return err + } + data, err := asks.EncodeCondition(c) + if err != nil { + return err + } + _, err = js.Publish(ctx, subject, data) + return err + } +} + +// watchAsks hands each key of the gate's bucket asks, with its revision, to put, and each key taken away to gone, until +// ctx ends or the watch fails. +func watchAsks(ctx context.Context, kv jetstream.KeyValue, put func(key string, rev uint64, value []byte), gone func(key string)) error { + w, err := kv.Watch(ctx, ">") + if err != nil { + return err + } + defer w.Stop() + for { + select { + case <-ctx.Done(): + return nil + case e, ok := <-w.Updates(): + if !ok { + return errors.New("the watch ended") + } + switch { + case e == nil: + case e.Operation() == jetstream.KeyValuePut: + put(e.Key(), e.Revision(), e.Value()) + default: + gone(e.Key()) + } + } + } +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go index 046b7b8..50ce770 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go @@ -16,6 +16,9 @@ import ( "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" ) // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the @@ -121,3 +124,275 @@ func TestTellerWaitsForTheGate(t *testing.T) { // 1240/AC2 t.Errorf("the teller stopped with %v, want a clean stop", err) } } + +// The principals of the composed accounts file for the teller's tests, with the permissions mesh-controller composes for +// them (internal/broker): a module's own state is the read grants of its bucket and `$KV..>` to write it +// (state.go stateGrants, readGrants), a read of another module's bucket is the read grants alone, and a module that uses +// the seat operator-channel submits its own `accept.ask.`, `accept.cancel.` and `accept.condition.`, +// hears its own `event.decided.` and reads `$KV.asks..>` of the router's record (seattraffic.go +// SeatTrafficOf, with the messenger's seat declaring accepts ask, cancel, condition, emits decided, by-caller and records +// asks). The gate is a bundle on the tool runner's account (ADR 0437 §3), so the gate's writes are the account node-tools'. +// The permissions are these lists written out, not the controller's function: the broker tests of issue 1243 prove that. +func readOf(bucket string) []string { + stream := "KV_" + bucket + return []string{"$JS.API.STREAM.INFO." + stream, "$JS.API.DIRECT.GET." + stream + ".>", "$JS.API.CONSUMER.CREATE." + stream + ".>", + "$JS.API.CONSUMER.DELETE." + stream + ".>", "$JS.FC." + stream + ".>"} +} + +func keepsOf(bucket string) []string { return append(readOf(bucket), "$KV."+bucket+".>") } + +func user(name string, pub, sub []string) string { + q := func(l []string) string { + var b strings.Builder + for _, s := range l { + b.WriteString(fmt.Sprintf(" %q\n", s)) + } + return b.String() + } + return fmt.Sprintf(` { user: %q, password: "pw" + permissions { + publish { allow: [ +%s ] } + subscribe { allow: [ +%s ] } + allow_responses: true + } + } +`, name, q(pub), q(append([]string{"_INBOX." + name + ".>"}, sub...))) +} + +const ( + tellerAnswers = "mesh-delivery-teller_answers" + tellerAsked = "mesh-delivery-teller_asked" + gateAsksKV = "mesh-delivery-gate_asks" + seatAsk = "mesh.seat.operator-channel.accept.ask.mesh-delivery-teller" + seatCancel = "mesh.seat.operator-channel.accept.cancel.mesh-delivery-teller" + seatCondition = "mesh.seat.operator-channel.accept.condition.mesh-delivery-teller" + seatDecided = "mesh.seat.operator-channel.event.decided.mesh-delivery-teller" +) + +func fullConf(dir string) string { + teller := append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), readOf(gateAsksKV)...), + seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_asks.$KV.asks.mesh-delivery-teller.>") + view := append(readOf(tellerAnswers), readOf(gateAsksKV)...) + tools := append(keepsOf(gateAsksKV), readOf(tellerAnswers)...) + return fmt.Sprintf(` +listen: 127.0.0.1:-1 +jetstream { store_dir: %q } +accounts { + MESH { + jetstream: enabled + users = [ + { user: "ctl", password: "pw" } +%s%s%s ] + } +} +`, filepath.Join(dir, "js"), + user("n1.mesh-delivery-teller", teller, []string{seatDecided}), + user("n1.mesh-delivery-view", view, nil), + user("n1.node-tools", tools, nil)) +} + +type fullBed struct { + srv *server.Server + ctl *nats.Conn + js jetstream.JetStream + ctx context.Context +} + +// startFull starts the bus with the principals above and what the controller makes before any module starts: the buckets +// of the gate's asks, the teller's answers and asked, the router's work queue SEAT_OPERATOR_CHANNEL. +func startFull(t *testing.T) *fullBed { + t.Helper() + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "t") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + conf := filepath.Join(dir, "bus.conf") + if err := os.WriteFile(conf, []byte(fullConf(dir)), 0o600); err != nil { + t.Fatal(err) + } + opts, err := server.ProcessConfigFile(conf) + if err != nil { + t.Fatal(err) + } + opts.NoLog, opts.NoSigs = true, true + srv, err := server.NewServer(opts) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(ctl.Close) + js, err := jetstream.New(ctl) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + t.Cleanup(cancel) + for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1} { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil { + t.Fatal(err) + } + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEAT_OPERATOR_CHANNEL", Subjects: []string{"mesh.seat.operator-channel.accept.>"}, + Retention: jetstream.WorkQueuePolicy}); err != nil { + t.Fatal(err) + } + return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx} +} + +func (b *fullBed) as(t *testing.T, name string) jetstream.JetStream { + t.Helper() + c, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo(name, "pw"), nats.CustomInboxPrefix("_INBOX."+name)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(c.Close) + js, err := jetstream.New(c) + if err != nil { + t.Fatal(err) + } + return js +} + +// put writes a key as a principal, within a bounded time: a write the permissions refuse reaches the writer as a timeout. +func (b *fullBed) put(t *testing.T, js jetstream.JetStream, bucket, key string) error { + t.Helper() + ctx, cancel := context.WithTimeout(b.ctx, 3*time.Second) + defer cancel() + kv, err := js.KeyValue(ctx, bucket) + if err != nil { + t.Fatalf("%s cannot be bound: %v", bucket, err) + } + _, err = kv.Put(ctx, key, []byte(`{}`)) + return err +} + +// TestCredentialsWriteOnlyTheirOwnBuckets: the teller's credential is refused writing the gate's asks, and the gate's +// (the tool runner's) and the view's are refused writing the teller's answers; each writes where it may, so a refusal is the +// permissions' and not a bus that does not take writes (1343/AC4, ADR 0437 §6's check). +func TestCredentialsWriteOnlyTheirOwnBuckets(t *testing.T) { // 1343/AC4 + b := startFull(t) + teller, view, tools := b.as(t, "n1.mesh-delivery-teller"), b.as(t, "n1.mesh-delivery-view"), b.as(t, "n1.node-tools") + if err := b.put(t, teller, tellerAnswers, "merge.g.0123456789ab"); err != nil { + t.Errorf("the teller cannot write its own answers: %v", err) + } + if err := b.put(t, tools, gateAsksKV, "merge.g.0123456789ab"); err != nil { + t.Errorf("the gate's account cannot write the gate's asks: %v", err) + } + if err := b.put(t, teller, gateAsksKV, "merge.g.ffffffffffff"); err == nil { + t.Error("the teller's credential wrote the gate's asks") + } + if err := b.put(t, tools, tellerAnswers, "merge.g.ffffffffffff"); err == nil { + t.Error("the gate's account wrote the teller's answers") + } + if err := b.put(t, view, tellerAnswers, "merge.g.ffffffffffff"); err == nil { + t.Error("the view's credential wrote the teller's answers") + } + // What was refused is not there. + for _, c := range []struct{ bucket, key string }{{gateAsksKV, "merge.g.ffffffffffff"}, {tellerAnswers, "merge.g.ffffffffffff"}} { + kv, _ := b.js.KeyValue(b.ctx, c.bucket) + if _, err := kv.Get(b.ctx, c.key); err == nil { + t.Errorf("%s holds %s though its write was refused", c.bucket, c.key) + } + } +} + +// TestTellerCarriesAnAskAndWritesTheAnswer starts the program on the composed principal against the bus: the gate's account +// writes an ask, the teller publishes it to the seat's work queue under its own name, the router (the test) says the +// warrant on the teller's decided subject, and the answer is under the ask's key in the teller's bucket; withdrawing a +// second ask cancels it (1343/AC1, AC2). +func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2 + now = time.Now().UTC() + b := startFull(t) + tools := b.as(t, "n1.node-tools") + broker := filepath.Join(t.TempDir(), "broker") + cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) + if err := os.WriteFile(broker, cred, 0o600); err != nil { + t.Fatal(err) + } + env := map[string]string{"MESH_BROKER_FILE": broker} + logs := &syncBuffer{} + rn := startRunner(b.ctx, func(c context.Context) error { return run(c, func(k string) string { return env[k] }, logs) }) + t.Cleanup(func() { rn.finish(t, 10*time.Second) }) + + key, value, ga := gateAsk(t, twoHeads(), twoHeads()) + kv, err := tools.KeyValue(b.ctx, gateAsksKV) + if err != nil { + t.Fatal(err) + } + // The program hears warrants before it reads the asks; give it its subscription before the ask is written. + time.Sleep(500 * time.Millisecond) + if _, err := kv.Put(b.ctx, key, value); err != nil { + t.Fatal(err) + } + seat, err := b.js.Stream(b.ctx, "SEAT_OPERATOR_CHANNEL") + if err != nil { + t.Fatal(err) + } + var raw *jetstream.RawStreamMsg + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if raw, err = seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil { + break + } + } + if raw == nil { + t.Fatalf("no ask on the seat's queue: %v\n%s", err, logs.String()) + } + var asked asks.Ask + if err := json.Unmarshal(raw.Data, &asked); err != nil || asked.Digest() != ga.Ask.Digest() { + t.Fatalf("the ask on the queue is %+v (%v), want the gate's", asked, err) + } + // The router says the warrant. + w, _ := json.Marshal(warrantFor(ga.Ask, "approve")) + if err := b.ctl.Publish(seatDecided, w); err != nil { + t.Fatal(err) + } + answers, err := b.js.KeyValue(b.ctx, tellerAnswers) + if err != nil { + t.Fatal(err) + } + var e jetstream.KeyValueEntry + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err = answers.Get(b.ctx, delivery.AnswerKey(key)); err == nil { + break + } + } + if e == nil { + t.Fatalf("no answer under %s: %v\n%s", key, err, logs.String()) + } + if a, err := delivery.DecodeAnswer(e.Value()); err != nil || a.Option != "approve" { + t.Errorf("answer %+v err %v", a, err) + } + // A second ask, withdrawn by the gate, is cancelled at the router. + other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}} + key2, value2, ga2 := gateAsk(t, other, other) + if _, err := kv.Put(b.ctx, key2, value2); err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if _, err := seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil && strings.Contains(logs.String(), ga2.Ask.ID) { + break + } + } + if err := kv.Delete(b.ctx, key2); err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if raw, err = seat.GetLastMsgForSubject(b.ctx, seatCancel); err == nil { + break + } + } + if err != nil || string(raw.Data) != `{"id":"`+ga2.Ask.ID+`"}` { + t.Errorf("the withdrawn ask was not cancelled: %v %v\n%s", raw, err, logs.String()) + } +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go new file mode 100644 index 0000000..8301912 --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go @@ -0,0 +1,101 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// kvStore is a state bucket of the teller as the asker's store (mesh-sdk go/asker Store): a compare-and-set that lost is +// stdio.ErrChanged, as the runtime's state reports it. +type kvStore struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (s kvStore) Get(key string) (*stdio.StateEntry, error) { + e, err := s.kv.Get(s.ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &stdio.StateEntry{Key: key, Value: e.Value(), Revision: e.Revision()}, nil +} + +func (s kvStore) Create(key string, value any) (uint64, error) { + b, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := s.kv.Create(s.ctx, key, b) + if errors.Is(err, jetstream.ErrKeyExists) { + return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) + } + return rev, err +} + +func (s kvStore) Update(key string, value any, revision uint64) (uint64, error) { + b, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := s.kv.Update(s.ctx, key, b, revision) + if errors.Is(err, jetstream.ErrKeyExists) { + return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) + } + return rev, err +} + +func (s kvStore) Delete(key string) error { + err := s.kv.Delete(s.ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil + } + return err +} + +// kvAnswers is the bucket answers: the teller's, written by compare-and-set only when the key has no answer. +type kvAnswers struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (a kvAnswers) Create(key string, value []byte) (bool, error) { + _, err := a.kv.Create(a.ctx, key, value) + if errors.Is(err, jetstream.ErrKeyExists) { + return false, nil + } + return err == nil, err +} + +func (a kvAnswers) Has(key string) (bool, error) { + _, err := a.kv.Get(a.ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return false, nil + } + return err == nil, err +} + +// kvGate reads the gate's bucket asks, which the teller may watch and read and never write. +type kvGate struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (g kvGate) Get(key string) ([]byte, bool, error) { + e, err := g.kv.Get(g.ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, false, nil + } + if err != nil { + return nil, false, err + } + return e.Value(), true, nil +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index 8005534..f9f3479 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -1,8 +1,9 @@ -// Command mesh-delivery-teller is delivery's teller (novox/hq ADR 0437 §4, issue 1240). It will carry the gate's asks to -// the operator and write the answers back (release 3d). In release 3c nothing writes asks, so it only runs: it connects -// as itself, says once that it waits for the gate's asks, and writes and publishes nothing. It does not open the asks -// bucket, for waiting for an absent bucket needs a timer (ADR 0437 §8 lets the parts hold none); the watch comes with the -// gate. It serves no request subject. +// Command mesh-delivery-teller is delivery's teller (novox/hq ADR 0437 §4, §6, §9; issues 1240, 1343). It carries the +// gate's merge asks to the operator and writes the answers back: it watches the gate's asks, asks each of the router as +// itself, and on the router's warrant, once it holds, writes the answer under the ask's key. Where the gate's asks +// bucket does not exist yet it says once that it waits for the gate's asks and does nothing: waiting for an absent bucket +// needs a timer (ADR 0437 §8 lets the parts hold none), so it is started again with the gate. It serves no request +// subject. package main import ( @@ -22,6 +23,10 @@ import ( "syscall" "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/asker" + "git.novox.be/novox/mesh-sdk/go/delivery" ) func main() { @@ -103,10 +108,43 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return fmt.Errorf("connecting to the bus: %w", err) } defer conn.Close() - logger.Print(waiting) - <-ctx.Done() + js, err := jetstream.New(conn) + if err != nil { + return err + } + gate, err := js.KeyValue(ctx, delivery.AsksBucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + logger.Print(waiting) + <-ctx.Done() + return nil + } + if err != nil { + return fmt.Errorf("the bucket %s: %w", delivery.AsksBucket, err) + } + var kvs [2]jetstream.KeyValue + for i, name := range []string{delivery.AnswersBucket, askedBucket} { + if kvs[i], err = js.KeyValue(ctx, name); err != nil { + return fmt.Errorf("the bucket %s: %w", name, err) + } + } + state := kvStore{ctx: ctx, kv: kvs[1]} + client := &asker.Client{Module: delivery.TellerModule, State: state, Publish: publishVia(ctx, js), + Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(conn, delivery.TellerModule, logger.Printf)} + t := &Teller{Asker: client, Store: state, Answers: kvAnswers{ctx: ctx, kv: kvs[0]}, Gate: kvGate{ctx: ctx, kv: gate}, + Say: sayVia(ctx, js, delivery.TellerModule), Logf: logger.Printf} + // Hear the warrants first, so none that comes while the asks are read is missed. + if err := client.Listen(t.Answer, t.Said); err != nil { + return fmt.Errorf("hearing the router's warrants: %w", err) + } + if err := watchAsks(ctx, gate, t.OnAsk, t.OnAskGone); err != nil { + return fmt.Errorf("watching the gate's asks: %w", err) + } return nil } +// askedBucket is the teller's state bucket asked, `_`: what it asked of the router and which condition it +// raised (manifest state entry asked, novox/hq ADR 0437 §4). +const askedBucket = delivery.TellerModule + "_asked" + // waiting is said once at start. -const waiting = "waiting for the gate's asks, which come with the gate in release 3d; nothing is read, written or published until then" +const waiting = "waiting for the gate's asks, whose bucket does not exist yet; the teller is started again with the gate, and nothing is read, written or published until then" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go new file mode 100644 index 0000000..921961f --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go @@ -0,0 +1,383 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asker" + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The teller carries the gate's merge asks to the operator and writes the operator's answer back (novox/hq ADR 0437 +// §6, §9). It reads the gate's asks, publishes each to the router unchanged as the asker mesh-delivery-teller +// (mesh-sdk go/asker), withdraws it when the gate takes its key away, and on the warrant the router says to it +// holds the warrant to the ask (ADR 0259 §6) and writes the answer under the ask's own key by compare-and-set. A +// warrant that does not hold writes no answer and is said as a condition. It holds no timer and serves no request: +// every reaction is a revision of the gate's key or a warrant. + +// store is where the teller keeps what it asked: the asker's own records (`ask.`) and, beside them, which gate key +// each ask was made for and at which revision of it (`key.`), and which conditions it has raised (`cond.`). +type store interface { + asker.Store + Delete(key string) error +} + +// routerAsks is the asker.Client as the teller uses it. +type routerAsks interface { + Ask(a asks.Ask) error + Cancel(id string) error + Where(id string) (state string, w *asks.Warrant, err error) + Take(w asks.Warrant, act func(asks.Option, asks.Warrant) error) (asker.Handled, error) +} + +// answers is the teller's bucket answers. +type answers interface { + // Create writes the answer under the ask's key only when there is none; created is false when there is one. + Create(key string, value []byte) (created bool, err error) + // Has says whether there is an answer under the key. + Has(key string) (bool, error) +} + +// gateAsks reads the gate's bucket asks. +type gateAsks interface { + Get(key string) (value []byte, found bool, err error) +} + +// Teller is the teller's logic; the program's wiring is main.go and decided.go. +type Teller struct { + Asker routerAsks + Store store + Answers answers + Gate gateAsks + // Say publishes a condition to the operator (mesh.seat.operator-channel.accept.condition.). + Say func(c asks.Condition) error + Logf func(format string, args ...any) + + mu sync.Mutex +} + +func (t *Teller) logf(format string, args ...any) { + if t.Logf != nil { + t.Logf(format, args...) + } +} + +// seen is what the teller keeps beside an ask it made: the gate key it was made for and the revision of that key it was +// made at, so a replay of the revision asks nothing again. +type seen struct { + Key string `json:"key"` + Rev uint64 `json:"rev"` +} + +func recordKey(id string) string { return "key." + id } +func keptKey(id string) string { return "ask." + id } +func condKey(id string) string { return "cond." + id } + +func (t *Teller) seenOf(id string) (seen, *stdio.StateEntry, error) { + e, err := t.Store.Get(recordKey(id)) + if err != nil || e == nil { + return seen{}, nil, err + } + var s seen + if err := json.Unmarshal(e.Value, &s); err != nil { + return seen{}, nil, fmt.Errorf("the record %s cannot be read: %w", recordKey(id), err) + } + return s, e, nil +} + +func (t *Teller) keptOf(id string) (*asker.Kept, uint64, error) { + e, err := t.Store.Get(keptKey(id)) + if err != nil || e == nil { + return nil, 0, err + } + var k asker.Kept + if err := json.Unmarshal(e.Value, &k); err != nil { + return nil, 0, fmt.Errorf("the record %s cannot be read: %w", keptKey(id), err) + } + return &k, e.Revision, nil +} + +// open says whether an ask is made and neither answered nor ended. +func open(k *asker.Kept) bool { return k != nil && !k.Acted && k.Ended == "" } + +func (t *Teller) note(id, key string, rev uint64, prior *stdio.StateEntry) error { + s := seen{Key: key, Rev: rev} + if prior == nil { + _, err := t.Store.Create(recordKey(id), s) + return err + } + _, err := t.Store.Update(recordKey(id), s, prior.Revision) + return err +} + +// OnAsk handles the gate's asks key put at revision rev: the ask it holds is published to the router once per +// revision of its key. A key whose value is not the ask its name says, one already answered, and a revision the teller +// has handled are left. An ask still open and unchanged is only noted (and its warrant read from the router, in case it +// came while the teller was away); one whose words changed is cancelled and asked afresh; one that ended, or that the +// teller cancelled, is asked again under the same id when the gate writes its key again (ADR 0441 §1). +func (t *Teller) OnAsk(key string, rev uint64, value []byte) { + t.mu.Lock() + defer t.mu.Unlock() + ga, err := delivery.DecodeAsk(value) + if err != nil { + t.logf("the gate's key %s at revision %d is not an ask, and is left: %v", key, rev, err) + return + } + if want, err := delivery.MergeAskKey(ga.Group, ga.Heads); err != nil || want != key { + t.logf("the gate's key %s at revision %d holds the ask of %q, not its own, and is left (%v)", key, rev, want, err) + return + } + id := ga.Ask.ID + if has, err := t.Answers.Has(key); err != nil { + t.logf("%s: the answers could not be read, so the ask is left: %v", key, err) + return + } else if has { + return + } + prior, srec, err := t.seenOf(id) + if err != nil { + t.logf("%s: %v", key, err) + return + } + k, krev, err := t.keptOf(id) + if err != nil { + t.logf("%s: %v", key, err) + return + } + if srec != nil && rev <= prior.Rev { + if open(k) { + t.catchUp(id) + } + return + } + if k != nil { + if open(k) && k.Ask.Digest() == ga.Ask.Digest() { + _ = t.note(id, key, rev, srec) + t.catchUp(id) + return + } + if open(k) { + if err := t.Asker.Cancel(id); err != nil { + t.logf("%s: the ask %s changed and could not be cancelled, so it is left: %v", key, id, err) + return + } + t.logf("%s: the ask %s changed at revision %d: the old one is cancelled and the new one asked", key, id, rev) + } + if err := t.Store.Delete(keptKey(id)); err != nil { + t.logf("%s: the earlier ask %s (revision %d of its record) could not be forgotten: %v", key, id, krev, err) + return + } + } + if err := t.note(id, key, rev, srec); err != nil { + t.logf("%s: the record of the ask %s could not be kept, so it is left: %v", key, id, err) + return + } + if err := t.Asker.Ask(ga.Ask); err != nil { + t.logf("%s at revision %d: the ask %s was not asked: %v", key, rev, id, err) + if kept, _, _ := t.keptOf(id); open(kept) { + // Kept but not published: forgotten, so the next revision of the key asks it. + _ = t.Store.Delete(keptKey(id)) + } + return + } + t.logf("%s at revision %d: the ask %s is asked of the operator", key, rev, id) +} + +// OnAskGone handles the gate's asks key taken away (a withdrawn ask): the ask is cancelled at the router, and an answer +// to it merges nothing. +func (t *Teller) OnAskGone(key string) { + t.mu.Lock() + defer t.mu.Unlock() + if _, _, _, err := delivery.ParseAskKey(key); err != nil { + t.logf("the gate's key %s was taken away and is no ask key, and is left: %v", key, err) + return + } + id := delivery.AskID(key) + k, krev, err := t.keptOf(id) + if err != nil { + t.logf("%s: %v", key, err) + return + } + if !open(k) { + return + } + if err := t.Asker.Cancel(id); err != nil { + t.logf("%s: the withdrawn ask %s could not be cancelled: %v", key, id, err) + return + } + // The router's word that it ended may come later: until it does, the ask is not open here. + k.Ended = asks.OutcomeCancelled + if _, err := t.Store.Update(keptKey(id), k, krev); err != nil && !errors.Is(err, stdio.ErrChanged) { + t.logf("%s: the cancelled ask %s could not be marked: %v", key, id, err) + } + t.logf("%s: the gate withdrew the ask %s: cancelled", key, id) +} + +// catchUp reads the router's record of an ask the teller made and hands its warrant on, as the router's event would: +// a warrant that came while the teller was away. +func (t *Teller) catchUp(id string) { + state, w, err := t.Asker.Where(id) + if err != nil { + t.logf("the router's record of the ask %s could not be read: %v", id, err) + return + } + if w == nil { + return + } + t.logf("the ask %s is %s at the router: its warrant is handled now", id, state) + h, err := t.Asker.Take(*w, t.Answer) + t.Said(h, *w, err) +} + +// mergeAct is what the approve option of the gate's ask binds: this ask, the answer, and the heads it covers. It is +// spelled as the gate composes it (mesh-catalog modules/mesh-delivery-gate approval.go mergeAct, from +// modules/mesh-delivery approval.go:63-65) and as mesh-controller verifies it (cmd/mesh-controller/moved.go:98-106). +func mergeAct(ask, option string, heads map[string]string) asks.Act { + var out []string + for id, sha := range heads { + out = append(out, id+"="+sha) + } + sort.Strings(out) + return asks.Act{"verb": "mesh-delivery.merge", "ask": ask, "option": option, "heads": strings.Join(out, ",")} +} + +// headsOf is the heads an ask covers as an act names them: a delivery's id, `owner/repository@` (mesh-catalog modules/mesh-delivery model.go Delivery.ID), and its whole commit. +func headsOf(heads []delivery.Head) map[string]string { + out := map[string]string{} + for _, h := range heads { + short := h.Head + if len(short) > 12 { + short = short[:12] + } + out[h.Repository+"@"+short] = h.Head + } + return out +} + +// warrantID names a warrant, which carries no id of its own: the ask and the time the router gave it. +func warrantID(w asks.Warrant) string { + return w.Ask + "@" + w.At.UTC().Format(time.RFC3339Nano) +} + +// Answer is the act of a warrant that holds the ask (asker.Client.Take has checked its asker, the ask's digest, the +// option and its level, and the time): the option's bound act must be the act of this ask and the heads it covers, +// and then the answer is written under the ask's key, once. An error is a warrant that does not hold here. +func (t *Teller) Answer(o asks.Option, w asks.Warrant) error { + s, srec, err := t.seenOf(w.Ask) + if err != nil { + return err + } + if srec == nil { + return fmt.Errorf("the teller holds no gate key for the ask %s", w.Ask) + } + raw, found, err := t.Gate.Get(s.Key) + if err != nil { + return fmt.Errorf("the gate's ask %s could not be read: %w", s.Key, err) + } + if !found { + t.logf("%s: the ask %s was answered and the gate has taken it away: an answer to a withdrawn ask merges nothing", s.Key, w.Ask) + return nil + } + ga, err := delivery.DecodeAsk(raw) + if err != nil { + return fmt.Errorf("the gate's ask %s is not an ask: %w", s.Key, err) + } + if ga.Ask.ID != w.Ask || ga.Ask.Digest() != w.AskDigest { + return fmt.Errorf("the gate's ask %s is no longer the ask the operator answered (its digest is %s, the warrant's %s)", + s.Key, ga.Ask.Digest(), w.AskDigest) + } + if err := o.Performs(mergeAct(w.Ask, o.ID, headsOf(ga.Heads))); err != nil { + return err + } + body, err := delivery.EncodeAnswer(delivery.Answer{Option: o.ID, Warrant: warrantID(w), Words: w.Says()}) + if err != nil { + return err + } + created, err := t.Answers.Create(delivery.AnswerKey(s.Key), body) + if err != nil { + return fmt.Errorf("the answer under %s could not be written: %w", s.Key, err) + } + if !created { + t.logf("%s: an answer is there already, and is not written again", s.Key) + return nil + } + t.logf("%s: %s", s.Key, w.Says()) + return nil +} + +// Said is what the teller does with what became of a warrant: a good answer clears the condition of its ask; a warrant +// that does not hold, or that the router refused, is said as a condition (ADR 0437 §9). Nothing is written for either. +func (t *Teller) Said(h asker.Handled, w asks.Warrant, err error) { + switch { + case h == asker.Acted && err == nil: + t.logf("the warrant of %s was %s", w.Ask, h) + t.clear(w.Ask) + case h == asker.Refused || (h == asker.Acted && err != nil): + t.logf("the warrant of %s does not hold, and no answer is written: %v", w.Ask, err) + t.raise(w.Ask, "A merge Approve did not hold", "its answer did not hold here: "+oneLine(err)) + case h == asker.Recorded && w.Outcome == asks.OutcomeRefused: + t.logf("the router refused the ask %s: %s", w.Ask, w.Words) + t.raise(w.Ask, "A merge ask was refused", "the router refused it: "+w.Words) + case err != nil: + t.logf("the warrant of %s: %v", w.Ask, err) + } +} + +func oneLine(err error) string { + if err == nil { + return "no reason was given" + } + s := strings.Join(strings.Fields(err.Error()), " ") + if r := []rune(s); len(r) > 200 { + return string(r[:199]) + "…" + } + return s +} + +// conditionKey is the key of the condition about an ask. +func conditionKey(id string) string { return delivery.TellerModule + ".merge-ask." + id } + +func (t *Teller) raise(id, headline, why string) { + if t.Say == nil { + return + } + if _, err := t.Store.Create(condKey(id), seen{}); err != nil { + if errors.Is(err, stdio.ErrChanged) { + return // raised already + } + t.logf("the condition of %s could not be kept: %v", id, err) + } + needs := "to ask for the Approve again" + c := asks.Condition{Key: conditionKey(id), Severity: asks.SeverityWarning, Subject: asks.Scope{Scope: "merge", ID: id}, State: asks.StateRaised, + Summary: "The merge ask " + id + ": " + why, Headline: headline, Needs: needs, + Explanation: asks.NeedsYou + needs + ". The merge ask " + id + ", " + why + ". Nothing was merged.", + Resolved: "A good answer to the merge ask was written."} + if err := t.Say(c); err != nil { + t.logf("the condition of %s could not be said: %v", id, err) + _ = t.Store.Delete(condKey(id)) + } +} + +func (t *Teller) clear(id string) { + if t.Say == nil { + return + } + e, err := t.Store.Get(condKey(id)) + if err != nil || e == nil { + return + } + c := asks.Condition{Key: conditionKey(id), Subject: asks.Scope{Scope: "merge", ID: id}, State: asks.StateCleared} + if err := t.Say(c); err != nil { + t.logf("the condition of %s could not be cleared: %v", id, err) + return + } + _ = t.Store.Delete(condKey(id)) +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go new file mode 100644 index 0000000..4c5f18e --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go @@ -0,0 +1,375 @@ +package main + +import ( + "encoding/json" + "fmt" + "strings" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asker" + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +const ( + headA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + headB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" +) + +var now = time.Date(2026, 10, 20, 9, 0, 0, 0, time.UTC) + +// memStore is a state bucket in memory with the revisions and the compare-and-set of the runtime's state. +type memStore struct { + entries map[string]*stdio.StateEntry + seq uint64 +} + +func (m *memStore) Get(key string) (*stdio.StateEntry, error) { + if e, ok := m.entries[key]; ok { + c := *e + return &c, nil + } + return nil, nil +} + +func (m *memStore) put(key string, value any) (uint64, error) { + b, err := json.Marshal(value) + if err != nil { + return 0, err + } + m.seq++ + m.entries[key] = &stdio.StateEntry{Key: key, Value: b, Revision: m.seq} + return m.seq, nil +} + +func (m *memStore) Create(key string, value any) (uint64, error) { + if _, ok := m.entries[key]; ok { + return 0, stdio.ErrChanged + } + return m.put(key, value) +} + +func (m *memStore) Update(key string, value any, revision uint64) (uint64, error) { + if e, ok := m.entries[key]; !ok || e.Revision != revision { + return 0, stdio.ErrChanged + } + return m.put(key, value) +} + +func (m *memStore) Delete(key string) error { delete(m.entries, key); return nil } + +type memAnswers struct { + written map[string][]byte + creates int +} + +func (a *memAnswers) Create(key string, value []byte) (bool, error) { + a.creates++ + if _, ok := a.written[key]; ok { + return false, nil + } + a.written[key] = value + return true, nil +} + +func (a *memAnswers) Has(key string) (bool, error) { _, ok := a.written[key]; return ok, nil } + +type memGate map[string][]byte + +func (g memGate) Get(key string) ([]byte, bool, error) { v, ok := g[key]; return v, ok, nil } + +type published struct { + subject string + body json.RawMessage + id string +} + +// bed is a teller with the asker.Client of the SDK on in-memory ends: what is published, the router's records, the +// warrants it would hear, and the conditions it says. +type bed struct { + t *Teller + client *asker.Client + store *memStore + answers *memAnswers + gate memGate + pubs []published + conds []asks.Condition + records map[string]string + hear func(stdio.Envelope) error + logs strings.Builder +} + +func newBed(t *testing.T) *bed { + t.Helper() + b := &bed{store: &memStore{entries: map[string]*stdio.StateEntry{}}, answers: &memAnswers{written: map[string][]byte{}}, + gate: memGate{}, records: map[string]string{}} + b.client = &asker.Client{Module: delivery.TellerModule, State: b.store, Now: func() time.Time { return now }, + Publish: func(subject string, body any, id string) error { + raw, err := json.Marshal(body) + b.pubs = append(b.pubs, published{subject, raw, id}) + return err + }, + Record: func(id string) (*stdio.StateEntry, error) { + if v, ok := b.records[id]; ok { + return &stdio.StateEntry{Key: id, Value: json.RawMessage(v)}, nil + } + return nil, nil + }, + Subscribe: func(_ string, handle func(stdio.Envelope) error) error { b.hear = handle; return nil }, + } + b.t = &Teller{Asker: b.client, Store: b.store, Answers: b.answers, Gate: b.gate, + Say: func(c asks.Condition) error { + if _, err := asks.EncodeCondition(c); err != nil { + t.Errorf("a condition the router would refuse: %v", err) + } + b.conds = append(b.conds, c) + return nil + }, + Logf: func(f string, a ...any) { fmt.Fprintf(&b.logs, f+"\n", a...) }} + if err := b.client.Listen(b.t.Answer, b.t.Said); err != nil { + t.Fatal(err) + } + return b +} + +// gateAsk is an ask as the gate composes it: the router's ask with the approve and decline options binding the act +// `mesh-delivery.merge` for this ask and the heads (mesh-catalog modules/mesh-delivery approval.go:123-191 composeMergeAsk +// and :63-65 mergeAct), under the id its key gives (mesh-sdk go/delivery AskID). boundTo are the heads the options bind, +// which are the ask's own unless a test says another. +func gateAsk(t *testing.T, heads, boundTo []delivery.Head) (key string, value []byte, ask delivery.Ask) { + t.Helper() + key, err := delivery.MergeAskKey("g", heads) + if err != nil { + t.Fatal(err) + } + id := delivery.AskID(key) + a := asks.Ask{ID: id, Headline: "Merge into main: novox/mesh-delivery#12", Explanation: "Merges two pull requests.\nReview: approved.", + Who: asks.Operator, Expires: now.Add(asks.ApproveLasts - 10*time.Minute), OnExpiry: "the merge stays held", About: "delivery.mesh-delivery.12"} + for _, o := range []struct{ id, label, does string }{{"approve", "Approve", "merges"}, {"decline", "Decline", "merges nothing"}} { + binds, err := asks.ActDigest(mergeAct(id, o.id, headsOf(boundTo))) + if err != nil { + t.Fatal(err) + } + a.Options = append(a.Options, asks.Option{ID: o.id, Label: o.label, Does: o.does, Level: asks.Approve, Binds: binds}) + } + ask = delivery.Ask{Group: "g", Heads: heads, Ask: a} + value, err = delivery.EncodeAsk(ask) + if err != nil { + t.Fatal(err) + } + return key, value, ask +} + +func twoHeads() []delivery.Head { + return []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headA}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}} +} + +// warrantFor is the router's warrant for a chosen option of an ask, as the router verified the operator. +func warrantFor(a asks.Ask, option string) asks.Warrant { + return asks.Warrant{Ask: a.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeChosen, Option: option, Label: "Approve", Level: asks.Approve, + By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "42", Verified: "user id"}, Channel: "telegram", + At: now.Add(time.Hour), AskDigest: a.Digest()} +} + +func (b *bed) deliver(t *testing.T, w asks.Warrant) { + t.Helper() + raw, _ := json.Marshal(w) + if err := b.hear(stdio.Envelope{Key: asks.Seat + ".decided." + delivery.TellerModule, Body: raw}); err != nil { + t.Fatal(err) + } +} + +func (b *bed) askedOf(subject string) []published { + var out []published + for _, p := range b.pubs { + if p.subject == subject { + out = append(out, p) + } + } + return out +} + +func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1343/AC1 + b := newBed(t) + key, value, ga := gateAsk(t, twoHeads(), twoHeads()) + b.gate[key] = value + b.t.OnAsk(key, 3, value) + asked := b.askedOf(asks.AskSubject(delivery.TellerModule)) + if len(asked) != 1 || asked[0].id != ga.Ask.ID { + t.Fatalf("published %+v, want one ask under %s with id %s", b.pubs, delivery.TellerModule, ga.Ask.ID) + } + var got asks.Ask + if err := json.Unmarshal(asked[0].body, &got); err != nil { + t.Fatal(err) + } + if got.Digest() != ga.Ask.Digest() || got.Headline != ga.Ask.Headline || got.Explanation != ga.Ask.Explanation || got.ID != delivery.AskID(key) { + t.Errorf("the router's ask differs from the gate's: %+v", got) + } + b.t.OnAsk(key, 3, value) // the revision seen again, as at a restart + b.t.OnAsk(key, 4, value) // the key written again with the same ask + if n := len(b.askedOf(asks.AskSubject(delivery.TellerModule))); n != 1 { + t.Errorf("the ask was published %d times", n) + } + delete(b.gate, key) + b.t.OnAskGone(key) + cancels := b.askedOf(asks.CancelSubject(delivery.TellerModule)) + if len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+ga.Ask.ID+`"}` { + t.Fatalf("cancels = %+v", cancels) + } + b.t.OnAskGone(key) + if n := len(b.askedOf(asks.CancelSubject(delivery.TellerModule))); n != 1 { + t.Errorf("the ask was cancelled %d times", n) + } +} + +func TestAWarrantThatHoldsWritesOneAnswerAndASecondWritesNone(t *testing.T) { // 1343/AC2 + b := newBed(t) + key, value, ga := gateAsk(t, twoHeads(), twoHeads()) + b.gate[key] = value + b.t.OnAsk(key, 1, value) + w := warrantFor(ga.Ask, "approve") + b.deliver(t, w) + raw, ok := b.answers.written[delivery.AnswerKey(key)] + if !ok || len(b.answers.written) != 1 { + t.Fatalf("answers = %v", b.answers.written) + } + a, err := delivery.DecodeAnswer(raw) + if err != nil || a.Option != "approve" || a.Warrant != w.Ask+"@"+w.At.UTC().Format(time.RFC3339Nano) || !strings.Contains(a.Words, "chose Approve") { + t.Errorf("answer %+v err %v", a, err) + } + b.deliver(t, w) // the same warrant again + w2 := w + w2.At = w.At.Add(time.Minute) // a second warrant for the same ask + b.deliver(t, w2) + if b.answers.creates != 1 { + t.Errorf("the answer was written %d times, want once", b.answers.creates) + } + if len(b.conds) != 0 { + t.Errorf("conditions said: %+v", b.conds) + } +} + +func TestAWarrantThatDoesNotHoldWritesNoAnswerAndRaisesTheCondition(t *testing.T) { // 1343/AC3 + other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headB}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}} + cases := map[string]func(t *testing.T) (*bed, string, asks.Warrant){ + "heads that differ from the ones the option bound": func(t *testing.T) (*bed, string, asks.Warrant) { + b := newBed(t) + key, value, ga := gateAsk(t, twoHeads(), other) // the options bind other heads than the ask covers + b.gate[key] = value + b.t.OnAsk(key, 1, value) + return b, key, warrantFor(ga.Ask, "approve") + }, + "an asker that differs": func(t *testing.T) (*bed, string, asks.Warrant) { + b := newBed(t) + key, value, ga := gateAsk(t, twoHeads(), twoHeads()) + b.gate[key] = value + b.t.OnAsk(key, 1, value) + w := warrantFor(ga.Ask, "approve") + w.Asker = "mesh-delivery" + return b, key, w + }, + "a digest that differs": func(t *testing.T) (*bed, string, asks.Warrant) { + b := newBed(t) + key, value, ga := gateAsk(t, twoHeads(), twoHeads()) + b.gate[key] = value + b.t.OnAsk(key, 1, value) + w := warrantFor(ga.Ask, "approve") + w.AskDigest = "sha256:0000" + return b, key, w + }, + } + for name, mk := range cases { + t.Run(name, func(t *testing.T) { + b, key, w := mk(t) + b.deliver(t, w) + if len(b.answers.written) != 0 { + t.Errorf("an answer was written: %v", b.answers.written) + } + if len(b.conds) != 1 || b.conds[0].State != asks.StateRaised || b.conds[0].Subject.ID != delivery.AskID(key) || + !strings.HasPrefix(b.conds[0].Explanation, asks.NeedsYou) { + t.Fatalf("conditions = %+v", b.conds) + } + }) + } +} + +func TestAGoodAnswerClearsTheConditionOfAnAskAskedAgain(t *testing.T) { // 1343/AC3 (cleared when a good answer is written) + b := newBed(t) + other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headB}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}} + key, bad, gaBad := gateAsk(t, twoHeads(), other) + b.gate[key] = bad + b.t.OnAsk(key, 1, bad) + b.deliver(t, warrantFor(gaBad.Ask, "approve")) + if len(b.conds) != 1 || b.conds[0].State != asks.StateRaised { + t.Fatalf("conditions = %+v", b.conds) + } + // The gate writes the ask again, right this time, at a new revision of its key: asked again under the same id. + _, good, gaGood := gateAsk(t, twoHeads(), twoHeads()) + b.gate[key] = good + b.t.OnAsk(key, 2, good) + if n := len(b.askedOf(asks.AskSubject(delivery.TellerModule))); n != 2 { + t.Fatalf("the ask was published %d times, want again at the new revision", n) + } + b.deliver(t, warrantFor(gaGood.Ask, "approve")) + if _, ok := b.answers.written[delivery.AnswerKey(key)]; !ok { + t.Fatalf("no answer: %v\n%s", b.answers.written, b.logs.String()) + } + if len(b.conds) != 2 || b.conds[1].State != asks.StateCleared || b.conds[1].Key != b.conds[0].Key { + t.Errorf("conditions = %+v", b.conds) + } +} + +func TestAWarrantComingWhileTheTellerWasAwayIsReadFromTheRouter(t *testing.T) { + b := newBed(t) + key, value, ga := gateAsk(t, twoHeads(), twoHeads()) + b.gate[key] = value + b.t.OnAsk(key, 1, value) + w, _ := json.Marshal(warrantFor(ga.Ask, "approve")) + b.records[ga.Ask.ID] = `{"state":"chosen","warrant":` + string(w) + `}` + b.t.OnAsk(key, 1, value) // the teller starts again and reads the key at the same revision + if _, ok := b.answers.written[delivery.AnswerKey(key)]; !ok { + t.Errorf("no answer: %s", b.logs.String()) + } +} + +func TestAnAskWhoseWordsChangedIsCancelledAndAskedAgain(t *testing.T) { + b := newBed(t) + key, value, ga := gateAsk(t, twoHeads(), twoHeads()) + b.t.OnAsk(key, 1, value) + changed := ga + changed.Ask.Explanation = "Merges two pull requests, now with a third point.\nReview: approved." + v2, err := delivery.EncodeAsk(changed) + if err != nil { + t.Fatal(err) + } + b.t.OnAsk(key, 2, v2) + if n := len(b.askedOf(asks.CancelSubject(delivery.TellerModule))); n != 1 { + t.Errorf("cancels = %d", n) + } + if n := len(b.askedOf(asks.AskSubject(delivery.TellerModule))); n != 2 { + t.Errorf("asks = %d", n) + } +} + +func TestAKeyWhoseValueIsNotItsOwnAskIsLeft(t *testing.T) { + b := newBed(t) + _, value, _ := gateAsk(t, twoHeads(), twoHeads()) + b.t.OnAsk("merge.g.000000000000", 1, value) + b.t.OnAsk("merge.g.000000000000", 2, []byte(`{"v":1`)) + if len(b.pubs) != 0 { + t.Errorf("published %+v", b.pubs) + } +} + +func TestARefusedAskIsSaidAsACondition(t *testing.T) { // ADR 0437 §9 + b := newBed(t) + key, value, ga := gateAsk(t, twoHeads(), twoHeads()) + b.t.OnAsk(key, 1, value) + w := asks.Warrant{Ask: ga.Ask.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeRefused, Words: "no channel can send", At: now} + b.deliver(t, w) + if len(b.conds) != 1 || b.conds[0].State != asks.StateRaised || !strings.Contains(b.conds[0].Summary, "no channel can send") { + t.Errorf("conditions = %+v", b.conds) + } +} diff --git a/modules/mesh-delivery-teller/go.mod b/modules/mesh-delivery-teller/go.mod index afbeb06..c977aa1 100644 --- a/modules/mesh-delivery-teller/go.mod +++ b/modules/mesh-delivery-teller/go.mod @@ -3,6 +3,7 @@ module mesh-delivery-teller go 1.26.0 require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019230201-5786ba190c69 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-teller/go.sum b/modules/mesh-delivery-teller/go.sum index 98dc72c..fcb33b1 100644 --- a/modules/mesh-delivery-teller/go.sum +++ b/modules/mesh-delivery-teller/go.sum @@ -1,3 +1,5 @@ +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019230201-5786ba190c69 h1:kvtV/SPsdRVUL/BZ3xxGS6U8YrXy/t7TRBjVrD6VLw8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019230201-5786ba190c69/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-teller/manifest_test.go b/modules/mesh-delivery-teller/manifest_test.go index 7343f02..438964f 100644 --- a/modules/mesh-delivery-teller/manifest_test.go +++ b/modules/mesh-delivery-teller/manifest_test.go @@ -76,7 +76,9 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 if m.Module != "mesh-delivery-teller" { t.Errorf("module = %q", m.Module) } - if want := []state{{Name: "answers", History: 5}}; !reflect.DeepEqual(m.State, want) { + // answers is the teller's answer to each ask; asked is what it asked of the router, which condition it raised and which + // gate key and revision each ask was made at (issue 1343). + if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-delivery-gate.asks"}; !reflect.DeepEqual(sorted(m.Reads), want) { @@ -134,7 +136,10 @@ func TestManifestProcess(t *testing.T) { // 1240/AC1 } func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1240/AC3 (ADR 0437 §8) - forbidden := []string{"Subscribe(", "QueueSubscribe", "ChanSubscribe", "time.NewTicker", "time.Tick", "time.NewTimer", "time.AfterFunc", "time.After(", "time.Sleep"} + forbidden := []string{"Subscribe(", "QueueSubscribe", "ChanSubscribe", "Respond(", "RespondMsg(", "time.NewTicker", "time.Tick", "time.NewTimer", "time.AfterFunc", "time.After(", "time.Sleep"} + // The one subscription the teller makes is to its own `decided` event, the router's word on its asks (ADR 0259, ADR 0437 + // §6): an event it hears, not a request it answers. It is allowed in decided.go and there only on that subject. + const decidedFile = "decided.go" var files int for _, root := range []string{"cmd", "internal"} { err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { @@ -150,6 +155,13 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1240/AC3 (ADR return err } for _, f := range forbidden { + if f == "Subscribe(" && filepath.Base(path) == decidedFile { + if n := strings.Count(string(src), "conn.Subscribe("); n != 1 || !strings.Contains(string(src), "conn.Subscribe(asks.DecidedSubject(module),") || + strings.Count(string(src), f) != n { + t.Errorf("%s holds %d subscriptions, and not just one to asks.DecidedSubject(module)", path, strings.Count(string(src), f)) + } + continue + } if strings.Contains(string(src), f) { t.Errorf("%s holds %s: the teller serves no request subject and holds no timer", path, f) } diff --git a/modules/mesh-delivery-teller/module.json b/modules/mesh-delivery-teller/module.json index cfed5f1..3c616f5 100644 --- a/modules/mesh-delivery-teller/module.json +++ b/modules/mesh-delivery-teller/module.json @@ -3,7 +3,8 @@ "version": "1", "slug": "teller", "state": [ - { "name": "answers", "history": 5 } + { "name": "answers", "history": 5 }, + { "name": "asked", "history": 1 } ], "reads": [ "mesh-delivery-gate.asks"