TL1: the teller tells a pipeline's widening on every channel, once per commit
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
This commit is contained in:
@@ -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": "<module>.<branch> 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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"
|
||||
|
||||
|
||||
@@ -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...)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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": [
|
||||
|
||||
Reference in New Issue
Block a user