From a489c738b31d34c9843b28872ddbd2a44b20c48a Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:34:43 +0100 Subject: [PATCH] TL1: the teller tells a pipeline's widening on every channel, once per commit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The teller hears pipeline-ended, tells each end that carries widens through the seat operator-channel's notify with every-channel, one line per entry, and keeps the last commit said per module and watched branch in its state widens-said (read by teller_widens_said); registration-changed drops the branches no longer watched. Builds: design 47 G §71, §335, §514, §522 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-teller-tools/main.go | 47 +++++ .../mesh-delivery-teller-tools/main_test.go | 17 ++ .../cmd/mesh-delivery-teller/kv.go | 9 + .../cmd/mesh-delivery-teller/main.go | 14 +- .../cmd/mesh-delivery-teller/names.go | 2 + .../cmd/mesh-delivery-teller/widens.go | 188 ++++++++++++++++++ .../cmd/mesh-delivery-teller/widens_test.go | 176 ++++++++++++++++ modules/mesh-delivery-teller/go.mod | 3 +- modules/mesh-delivery-teller/manifest_test.go | 9 +- modules/mesh-delivery-teller/module.json | 15 +- 10 files changed, 469 insertions(+), 11 deletions(-) create mode 100644 modules/mesh-delivery-teller/cmd/mesh-delivery-teller/widens.go create mode 100644 modules/mesh-delivery-teller/cmd/mesh-delivery-teller/widens_test.go diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go index 07fe581b..7aeeab66 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go @@ -20,6 +20,9 @@ import ( "mesh-delivery-teller/internal/ends" ) +// widensSaidState is the teller's state widens-said (the program's own constant of it is in cmd/mesh-delivery-teller). +const widensSaidState = "widens-said" + // asker writes an ends value and waits for the teller's answer to it. type asker func(key string, value []byte) (ends.End, error) @@ -101,6 +104,49 @@ func AnswerTools(s answers) []stdio.Tool { }}} } +// WidensSaidTool reads the teller's state widens-said (design 47 G §335). +const WidensSaidTool = "teller_widens_said" + +// WidensSaidTools is the tool teller_widens_said over the state widens-said: the last commit the teller said a widening of +// for one module on one watched branch, or every one when none is named. +func WidensSaidTools(s answers) []stdio.Tool { + return []stdio.Tool{{Name: WidensSaidTool, + Description: "Read what the delivery teller has said of widenings: for each module and watched branch, the last commit it told the " + + "operator a widening of on every channel. Keys are the module and the branch, each escaped as a key's name is, joined by a dot. " + + "Name a key for one, or none for every one.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "key": map[string]any{"type": "string", "description": ". as the state keys it; none for every one"}}}, + Run: func(args map[string]any) (any, error) { + key, _ := args["key"].(string) + if key = strings.TrimSpace(key); key != "" { + e, err := s.Get(key) + if err != nil { + return nil, fmt.Errorf("%s cannot be read: %w", key, err) + } + if e == nil { + return map[string]any{"key": key, "found": false}, nil + } + return map[string]any{"key": key, "found": true, "commit": json.RawMessage(e.Value)}, nil + } + keys, err := s.Keys() + if err != nil { + return nil, fmt.Errorf("the keys cannot be listed: %w", err) + } + sort.Strings(keys) + out := make([]map[string]any, 0, len(keys)) + for _, k := range keys { + e, err := s.Get(k) + if err != nil { + return nil, fmt.Errorf("%s cannot be read: %w", k, err) + } + if e != nil { + out = append(out, map[string]any{"key": k, "commit": json.RawMessage(e.Value)}) + } + } + return map[string]any{"said": out}, nil + }}} +} + // askEnd writes the ends value at key and waits, on a watch of the key, at most bound for the teller's answer. func askEnd(s stdio.KeptState, bound time.Duration) asker { return func(key string, value []byte) (ends.End, error) { @@ -159,6 +205,7 @@ func main() { tools := append(Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now), views.StatusTools("mesh-delivery-teller", "teller_first_views", "teller_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) tools = append(tools, AnswerTools(stdio.State("answers"))...) + tools = append(tools, WidensSaidTools(stdio.State(widensSaidState))...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-teller-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main_test.go index ede71d7a..92acec4c 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main_test.go @@ -47,3 +47,20 @@ func TestTellerAnswersReadsOneAnswerOrEveryAnswer(t *testing.T) { t.Errorf("every answer: %s, %v", raw, err) } } + +// teller_widens_said reads the state widens-said: the commit last said for one module and branch, or every one. +func TestTellerWidensSaidReadsOneKeyOrEveryKey(t *testing.T) { + tool := WidensSaidTools(fakeAnswers{"nats.main": `"c1"`, "nats.next": `"c2"`})[0] + if tool.Name != "teller_widens_said" { + t.Fatalf("the tool is %s", tool.Name) + } + out, err := tool.Run(map[string]any{"key": "nats.main"}) + if err != nil || fmt.Sprint(out) != `map[commit:"c1" found:true key:nats.main]` { + t.Errorf("one key: %v, %v", out, err) + } + out, err = tool.Run(map[string]any{}) + raw, _ := json.Marshal(out) + if err != nil || string(raw) != `{"said":[{"commit":"c1","key":"nats.main"},{"commit":"c2","key":"nats.next"}]}` { + t.Errorf("every key: %s, %v", raw, err) + } +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go index 13f7ecdc..9da622da 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go @@ -62,6 +62,15 @@ func (s kvStore) Delete(key string) error { return err } +// Keys are the keys that stand. +func (s kvStore) Keys() ([]string, error) { + keys, err := s.kv.Keys(s.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return keys, err +} + // kvAnswers is the bucket answers: the teller's, written by compare-and-set: created when the key has no answer, updated at the revision read when it holds an earlier asking's. type kvAnswers struct { ctx context.Context 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 34ef4bc2..f338d719 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -60,9 +60,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return process.EssentialErr("teller", gateAsksBucket, "tell any of the gate's asks", err) } // The answers it writes and what it asked: its own work, so each stops the start, naming its grant. - var kvs [2]jetstream.KeyValue - for i, name := range []string{answersBucket, askedBucket} { - if kvs[i], err = process.EssentialBucket(ctx, js, "teller", name, "write an answer or remember what it asked"); err != nil { + var kvs [3]jetstream.KeyValue + for i, name := range []string{answersBucket, askedBucket, widensSaidBucket} { + if kvs[i], err = process.EssentialBucket(ctx, js, "teller", name, "write an answer, remember what it asked or remember what it said of a widening"); err != nil { return err } } @@ -92,6 +92,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro _, err := js.Publish(ctx, cred.EventSubject(event), body, jetstream.WithMsgID(id)) return err }} + // A pipeline's widening is said on every channel once, and what was said is kept (design 47 G §335). + widens := &Widens{State: kvStore{ctx: ctx, kv: kvs[2]}, Call: calls.Call, Logf: logger.Printf} + if err := hub.Handle(pipelineEndedSubject, widens.HearEnded); err != nil { + return fmt.Errorf("hearing %s: %w", pipelineEndedSubject, err) + } + if err := hub.Handle(registrationChangedSubject, widens.HearRegistration); err != nil { + return fmt.Errorf("hearing %s: %w", registrationChangedSubject, err) + } // 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) diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/names.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/names.go index e9c2d0e0..1a55331f 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/names.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/names.go @@ -17,6 +17,8 @@ const ( viewAnswersBucket = viewModule + "_answers" // answersBucket is the teller's own state answers, which the gate reads. answersBucket = tellerModule + "_answers" + // widensSaidBucket is the teller's own state widens-said, the last commit it said a widening of per module and branch. + widensSaidBucket = tellerModule + "_" + widensSaidState // assignerAsksBucket is the assigner's state asks (the first view of assigner_read_asks stands in for it). assignerAsksBucket = assignerModule + "_asks" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/widens.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/widens.go new file mode 100644 index 00000000..4c116f1d --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/widens.go @@ -0,0 +1,188 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + mpr "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// What the teller says of a widening (design 47 G §71, §335, §522): the runner says each pipeline's end as pipeline-ended, +// its field `widens` naming what the change widens, and the teller tells the operator on every channel through the seat +// operator-channel's verb notify (§514), one message per pipeline-ended that carries any, once per module, watched branch +// and commit. It keeps, in its state widens-said, one key per module and watched branch holding the last commit it said, +// replaced by the next and deleted when registration-changed says the branch is no longer watched. + +const ( + // widensSaidState is the teller's state of the commits it said a widening of; teller_widens_said reads it. + widensSaidState = "widens-said" + + // notifyKey is the seat verb the teller tells the operator through (manifest invokes). + notifyKey = "seat:operator-channel.notify" + // registrationsKey is the seat mesh-catalog's verb that answers a module's watched branches (manifest invokes). + registrationsKey = "seat:mesh-catalog.registrations" +) + +// The events the teller hears for it (manifest consumes). +const ( + pipelineEndedSubject = "mesh.seat." + mpr.Seat + ".event." + mpr.EventPipelineEnded + ".>" + registrationChangedSubject = meshcatalog.SubjectRegistrationChanged +) + +// widensState is the state widens-said as the teller keeps it. +type widensState interface { + Get(key string) (*stdio.StateEntry, error) + Create(key string, value any) (uint64, error) + Update(key string, value any, revision uint64) (uint64, error) + Delete(key string) error + Keys() ([]string, error) +} + +// Widens says a pipeline's widening once and keeps what it said. +type Widens struct { + State widensState + // Call is the bus call of a seat verb as the teller's own account (process.Conn.Call). + Call func(key string, body any) (json.RawMessage, error) + Logf func(format string, args ...any) +} + +// widensKey is the key of a module on a watched branch, each name escaped as a key's name is. +func widensKey(module, branch string) string { + return keys.EscapeName(module) + "." + keys.EscapeName(branch) +} + +// commit7 is the commit's first seven characters. +func commit7(commit string) string { + if len(commit) > 7 { + return commit[:7] + } + return commit +} + +// widensWords is the message of a pipeline's end: one line per entry, in the operator's words (design 47 G §522). +func widensWords(p mpr.PipelineEnded) string { + lines := make([]string, 0, len(p.Widens)) + for _, w := range p.Widens { + lines = append(lines, fmt.Sprintf("Widening: %s %s on %s widens %s %s.", p.Module, commit7(p.Commit), p.Branch, w.What, w.Name)) + } + return strings.Join(lines, "\n") +} + +// Said tells the operator the widening of a pipeline's end on every channel, once for its module, watched branch and +// commit, and keeps the commit; an end with no widening, or one whose commit is the last said, says nothing. +func (w *Widens) Said(p mpr.PipelineEnded) error { + if len(p.Widens) == 0 || p.Commit == "" { + return nil + } + key := widensKey(p.Module, p.Branch) + e, err := w.State.Get(key) + if err != nil { + return fmt.Errorf("the commit last said for %s on %s cannot be read: %w", p.Module, p.Branch, err) + } + if e != nil && strings.Trim(string(e.Value), `"`) == p.Commit { + return nil + } + words := widensWords(p) + // The message's key holds the commit, so a say that is made again after a failure to keep it replaces the message. + if _, err := w.Call(notifyKey, map[string]any{"key": "widens." + key + "." + commit7(p.Commit), "subject": p.Module, + "severity": "warning", "summary": words, "headline": words, "every-channel": true}); err != nil { + return fmt.Errorf("the widening of %s %s on %s is not told: %w", p.Module, commit7(p.Commit), p.Branch, err) + } + if e == nil { + _, err = w.State.Create(key, p.Commit) + } else { + _, err = w.State.Update(key, p.Commit, e.Revision) + } + if err != nil { + return fmt.Errorf("the widening of %s %s on %s is told but not kept: %w", p.Module, commit7(p.Commit), p.Branch, err) + } + return nil +} + +// Unwatched deletes what was kept for each branch of module the registration no longer watches. +func (w *Widens) Unwatched(module string, watched []string) error { + keep := map[string]bool{} + for _, b := range watched { + keep[widensKey(module, b)] = true + } + all, err := w.State.Keys() + if err != nil { + return err + } + prefix := keys.EscapeName(module) + "." + for _, k := range all { + if strings.HasPrefix(k, prefix) && !keep[k] { + if err := w.State.Delete(k); err != nil { + return err + } + } + } + return nil +} + +// HearEnded is the handler of pipeline-ended. +func (w *Widens) HearEnded(m jetstream.Msg) { + var p mpr.PipelineEnded + if err := json.Unmarshal(m.Data(), &p); err != nil { + w.logf("%s is not a pipeline end and is left: %v", m.Subject(), err) + _ = m.Ack() + return + } + w.finish(m, w.Said(p)) +} + +// HearRegistration is the handler of registration-changed: it asks the registration of the module and drops the branches it +// no longer watches. +func (w *Widens) HearRegistration(m jetstream.Msg) { + var ev meshcatalog.RegistrationChangedEvent + if err := json.Unmarshal(m.Data(), &ev); err != nil || ev.Module == "" { + w.logf("%s names no module and is left: %v", m.Subject(), err) + _ = m.Ack() + return + } + raw, err := w.Call(registrationsKey, map[string]any{"module": ev.Module}) + if err != nil { + w.finish(m, fmt.Errorf("the registration of %s cannot be read: %w", ev.Module, err)) + return + } + var ans meshcatalog.RegistrationsAnswer + if err := json.Unmarshal(raw, &ans); err != nil { + w.finish(m, fmt.Errorf("the registration of %s is not read: %w", ev.Module, err)) + return + } + for _, r := range ans.Registrations { + if r.Module == ev.Module { + w.finish(m, w.Unwatched(ev.Module, r.Branches)) + return + } + } + // A module with no registration watches no branch. + w.finish(m, w.Unwatched(ev.Module, nil)) +} + +// finish acknowledges the message when err is nil and asks it again when it is not. +func (w *Widens) finish(m jetstream.Msg, err error) { + if err == nil { + if aerr := m.Ack(); aerr != nil { + w.logf("%s was handled but could not be acknowledged: %v", m.Subject(), aerr) + } + return + } + w.logf("%v; the message is heard again", err) + if nerr := m.Nak(); nerr != nil && !errors.Is(nerr, jetstream.ErrMsgAlreadyAckd) { + w.logf("%s could not be given back: %v", m.Subject(), nerr) + } +} + +func (w *Widens) logf(format string, args ...any) { + if w.Logf != nil { + w.Logf(format, args...) + } +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/widens_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/widens_test.go new file mode 100644 index 00000000..dcab8f71 --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/widens_test.go @@ -0,0 +1,176 @@ +package main + +import ( + "encoding/json" + "errors" + "slices" + "sort" + "strings" + "testing" + + "github.com/nats-io/nats.go/jetstream" + + mpr "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// widensState is memStore with the delete and the keys the state widens-said needs. +type widensMem struct{ memStore } + +func (m *widensMem) Create(key string, value any) (uint64, error) { + if _, ok := m.entries[key]; ok { + return 0, stdio.ErrChanged + } + return m.put(key, value) +} +func (m *widensMem) Update(key string, value any, rev uint64) (uint64, error) { + if e, ok := m.entries[key]; !ok || e.Revision != rev { + return 0, stdio.ErrChanged + } + return m.put(key, value) +} +func (m *widensMem) Delete(key string) error { delete(m.entries, key); return nil } +func (m *widensMem) Keys() ([]string, error) { + var out []string + for k := range m.entries { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +// calls records the seat verbs the teller calls and answers the registrations. +type calls struct { + notified []map[string]any + fail error + branches []string +} + +func (c *calls) call(key string, body any) (json.RawMessage, error) { + switch key { + case notifyKey: + if c.fail != nil { + return nil, c.fail + } + c.notified = append(c.notified, body.(map[string]any)) + return json.RawMessage(`{}`), nil + case registrationsKey: + return json.Marshal(map[string]any{"registrations": []map[string]any{{"module": "nats", "branches": c.branches}}}) + } + return nil, errors.New("not called: " + key) +} + +func widenEnd(commit string, what ...string) mpr.PipelineEnded { + p := mpr.PipelineEnded{Module: "nats", Branch: "main", Commit: commit} + for _, w := range what { + p.Widens = append(p.Widens, mpr.Widening{What: w, Name: "name-of-" + strings.ReplaceAll(w, " ", "-")}) + } + return p +} + +// Design 47 G §71, §335, §522 (TL1): a pipeline's widening is said once on every channel, in the operator's words. +func TestAWideningIsSaidOnceOnEveryChannelAndKept(t *testing.T) { + c := &calls{} + w := &Widens{State: &widensMem{memStore{entries: map[string]*stdio.StateEntry{}}}, Call: c.call} + end := widenEnd("0123456789abcdef0123456789abcdef01234567", "grant", "serving part") + steps := []struct { + name string + end mpr.PipelineEnded + want int // messages said so far + }{ + {"the first end says one message", end, 1}, + {"the same end again says nothing", end, 1}, + {"an end with no widening says nothing", widenEnd("fffffffffffffffffffffffffffffffffffffff"), 1}, + {"the next commit says one more", widenEnd("89abcdef0123456789abcdef0123456789abcdef", "credential"), 2}, + } + for _, s := range steps { + if err := w.Said(s.end); err != nil { + t.Fatalf("%s: %v", s.name, err) + } + if len(c.notified) != s.want { + t.Fatalf("%s: %d messages, want %d", s.name, len(c.notified), s.want) + } + } + first := c.notified[0] + if first["every-channel"] != true { + t.Errorf("the message is not for every channel: %v", first) + } + wantWords := "Widening: nats 0123456 on main widens grant name-of-grant.\nWidening: nats 0123456 on main widens serving part name-of-serving-part." + if first["headline"] != wantWords { + t.Errorf("the words are %q, want %q", first["headline"], wantWords) + } + if k := first["key"]; k != "widens.nats.main.0123456" { + t.Errorf("the key is %v", k) + } + if e, _ := w.State.Get("nats.main"); e == nil || string(e.Value) != `"89abcdef0123456789abcdef0123456789abcdef"` { + t.Errorf("the state holds %v, want the last commit said", e) + } +} + +// A message that is not told keeps nothing, so the end is heard again and told. +func TestAWideningNotToldIsNotKept(t *testing.T) { + c := &calls{fail: errors.New("no channel")} + st := &widensMem{memStore{entries: map[string]*stdio.StateEntry{}}} + w := &Widens{State: st, Call: c.call} + end := widenEnd("0123456789abcdef0123456789abcdef01234567", "grant") + if err := w.Said(end); err == nil { + t.Fatal("a say that failed returned no error") + } + if len(st.entries) != 0 { + t.Fatalf("kept %v though it was not told", st.entries) + } + c.fail = nil + if err := w.Said(end); err != nil || len(c.notified) != 1 { + t.Fatalf("heard again: %v, %d messages", err, len(c.notified)) + } +} + +// fakeMsg is a jetstream message of a subject and a body that records how it was answered. +type fakeMsg struct { + jetstream.Msg + data []byte + acked bool + naked bool +} + +func (f *fakeMsg) Data() []byte { return f.data } +func (f *fakeMsg) Subject() string { return "s" } +func (f *fakeMsg) Ack() error { f.acked = true; return nil } +func (f *fakeMsg) Nak() error { f.naked = true; return nil } + +// A registration-changed drops the keys of the branches the registration no longer watches, and only of its module. +func TestARegistrationChangedDropsTheBranchesNoLongerWatched(t *testing.T) { + c := &calls{branches: []string{"main"}} + st := &widensMem{memStore{entries: map[string]*stdio.StateEntry{}}} + for _, k := range []string{"nats.main", "nats.next", "natsx.next", "other.next"} { + st.put(k, "c") + } + w := &Widens{State: st, Call: c.call} + m := &fakeMsg{data: []byte(`{"module":"nats","moved":[],"version":2,"warrant":"w"}`)} + w.HearRegistration(m) + keys, _ := st.Keys() + if want := []string{"nats.main", "natsx.next", "other.next"}; !slices.Equal(keys, want) { + t.Errorf("keys %v, want %v", keys, want) + } + if !m.acked || m.naked { + t.Errorf("acked %v naked %v", m.acked, m.naked) + } +} + +// A pipeline-ended whose say fails is given back, and one that is told is acknowledged. +func TestAPipelineEndedIsAcknowledgedOnlyWhenTold(t *testing.T) { + c := &calls{fail: errors.New("no channel")} + w := &Widens{State: &widensMem{memStore{entries: map[string]*stdio.StateEntry{}}}, Call: c.call} + body, _ := json.Marshal(widenEnd("0123456789abcdef0123456789abcdef01234567", "grant")) + m := &fakeMsg{data: body} + w.HearEnded(m) + if m.acked || !m.naked { + t.Errorf("a failed say: acked %v naked %v", m.acked, m.naked) + } + c.fail = nil + m = &fakeMsg{data: body} + w.HearEnded(m) + if !m.acked || m.naked { + t.Errorf("a told say: acked %v naked %v", m.acked, m.naked) + } +} diff --git a/modules/mesh-delivery-teller/go.mod b/modules/mesh-delivery-teller/go.mod index c41a8430..96a10779 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-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 @@ -47,7 +48,7 @@ require ( git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/containerimageregistryclient v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/containerimageregistry v0.0.0-00010101000000-000000000000 // indirect - git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 // indirect + git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/ociartifactregistry v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000 // indirect ) diff --git a/modules/mesh-delivery-teller/manifest_test.go b/modules/mesh-delivery-teller/manifest_test.go index aca44653..75c044ed 100644 --- a/modules/mesh-delivery-teller/manifest_test.go +++ b/modules/mesh-delivery-teller/manifest_test.go @@ -94,7 +94,7 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 // 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); ends is the ends asked by hand with teller_end_ask (issue 774). // first-views is where its first view of the assigner's asks stands and findings its condition past the bound (ADR 0615). - if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}, {Name: "ends", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { + if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}, {Name: "ends", History: 1}, {Name: "first-views", History: 1}, {Name: "widens-said", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } // The view's answers to the assigner's asks are read so the view's answer fills an ask that needs no proof (1297/B10); the @@ -108,11 +108,12 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 } } // The join: its first view asked again by hand is a hand act, recorded through the seat (views.StatusTools). - if want := []string{"mesh-delivery-assigner.assigner_read_asks", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(m.Invokes, want) { + if want := []string{"mesh-delivery-assigner.assigner_read_asks", "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.registrations", "seat:operator-channel.notify"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } + // design 47 G §335 (TL1): the widening of a pipeline end is told through the seat operator-channel and kept in widens-said. // design 47 G §387: the keeper's ask for the findings is heard, and the finding stream is declared (§383). - if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks"}; !reflect.DeepEqual(m.Consumes, want) { + if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks", "mesh-catalog.registration-changed", "mesh-pipeline-runner.pipeline-ended"}; !reflect.DeepEqual(m.Consumes, want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if want := []string{"operator-channel"}; !reflect.DeepEqual(sorted(m.Uses), want) { @@ -125,7 +126,7 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 t.Errorf("secrets-owner = %q", m.SecretsOwner) } // 774: teller_end_ask, served by its tools bundle through the state ends; the process serves none. - if want := []string{"teller_end_ask", "teller_answers", "teller_first_views", "teller_first_view_again"}; !reflect.DeepEqual(m.Tools, want) { + if want := []string{"teller_end_ask", "teller_answers", "teller_first_views", "teller_first_view_again", "teller_widens_said"}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } // Its two events: each answer it writes (ADR 0596 §2), and an assigner ask ended at the channel without an answer, heard by diff --git a/modules/mesh-delivery-teller/module.json b/modules/mesh-delivery-teller/module.json index 4c00a0d3..606bd8de 100644 --- a/modules/mesh-delivery-teller/module.json +++ b/modules/mesh-delivery-teller/module.json @@ -6,15 +6,20 @@ "teller_end_ask", "teller_answers", "teller_first_views", - "teller_first_view_again" + "teller_first_view_again", + "teller_widens_said" ], "invokes": [ "mesh-delivery-assigner.assigner_read_asks", - "seat:mesh-hand-acts.hand-act" + "seat:mesh-hand-acts.hand-act", + "seat:mesh-catalog.registrations", + "seat:operator-channel.notify" ], "consumes": [ "mesh-conditions.findings-asked", - "mesh-delivery-assigner.views.asks" + "mesh-delivery-assigner.views.asks", + "mesh-catalog.registration-changed", + "mesh-pipeline-runner.pipeline-ended" ], "emits": [ "answered", @@ -36,6 +41,10 @@ { "name": "first-views", "history": 1 + }, + { + "name": "widens-said", + "history": 1 } ], "reads": [