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 new file mode 100644 index 0000000..a7e035c --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go @@ -0,0 +1,109 @@ +// Command mesh-delivery-teller-tools serves the teller's recovery tool (issue 774): the runtime launches it for a call. +// teller_end_ask writes one value of the teller's state ends and waits for the teller's answer there. The teller is the +// other program: it serves no request subject (ADR 0437 §4). +package main + +import ( + "context" + "encoding/json" + "fmt" + "os" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + + "mesh-delivery-teller/internal/ends" +) + +// asker writes an ends value and waits for the teller's answer to it. +type asker func(key string, value []byte) (ends.End, error) + +// Tools are the teller's tools. +func Tools(ask asker, now func() time.Time) []stdio.Tool { + str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } + return []stdio.Tool{ + {Name: ends.Tool, + Description: "End one open ask the delivery teller holds at the operator channel, by hand, saying why: the teller " + + "cancels it at the router, which frees one of the three places the router gives the teller, or drops it from the " + + "teller's wait for room. For an ask whose question is gone but whose key was never taken away (a merge ask of a pull " + + "request merged by another path, a backup ask about a backup no longer failing), which otherwise holds its place " + + "until it lapses, so a newer ask is refused for want of room. Name the ask by its router id (as " + + "operator-channel's asks show it) or by the key of the gate's or the assigner's asks. Answers which ask it found, " + + "whether it ended it, and what it did; an ask answered, lapsed or ended already is left and said so.", + Input: map[string]any{"type": "object", "required": []string{"ask", "why"}, "properties": map[string]any{ + "ask": map[string]any{"type": "string", "description": "the ask's router id, or the key of the gate's or the assigner's asks it was asked for"}, + "why": map[string]any{"type": "string", "description": "why it is ended by hand, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + key, value, err := ends.New(str(args, "ask"), str(args, "why"), stdio.CallerOf(args), now()) + if err != nil { + return nil, err + } + e, err := ask(key, value) + if err != nil { + return nil, err + } + return e.Result, 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) { + answered := make(chan ends.End, 1) + var rev uint64 + var mu sync.Mutex + take := func(revision uint64, v []byte) { + mu.Lock() + mine := rev + mu.Unlock() + if mine == 0 || revision <= mine { + return + } + e, err := ends.Decode(v) + if err != nil || !e.Done() { + return + } + select { + case answered <- e: + default: + } + } + if err := s.Watch(key, func(c stdio.StateChange) error { + if c.Op == "put" { + take(c.Revision, c.Value) + } + return nil + }); err != nil { + return ends.End{}, fmt.Errorf("watching %s.%s: %w", ends.State, key, err) + } + r, err := s.Put(key, json.RawMessage(value)) + if err != nil { + return ends.End{}, fmt.Errorf("the end could not be written to %s: %w", ends.Bucket, err) + } + mu.Lock() + rev = r + mu.Unlock() + // The watch may have heard the answer before the revision was known: read it once too. + if e, err := s.Get(key); err == nil && e != nil { + take(e.Revision, e.Value) + } + wait, cancel := context.WithTimeout(context.Background(), bound) // a context's bound, no timer (ADR 0437 §8) + defer cancel() + select { + case e := <-answered: + return e, nil + case <-wait.Done(): + return ends.End{}, fmt.Errorf("the teller did not answer within %s: is it running? its answer will be in %s.%s", bound, ends.Bucket, key) + } + } +} + +func main() { + if err := stdio.Serve("", Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now)); 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/ends.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/ends.go new file mode 100644 index 0000000..2237778 --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/ends.go @@ -0,0 +1,92 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "slices" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "mesh-delivery-teller/internal/ends" +) + +// EndAsk ends one ask of the teller's by hand (the tool teller_end_ask, issue 774): an ask open at the router is cancelled +// there and marked ended here, and one waiting for room is dropped from the wait, so it is asked no more under that router +// id. The ask is named by its router id or by the key of the gate's or the assigner's asks it was asked for (then its latest +// asking). An ask that is neither open nor waiting is said so and left: it holds no place at the router. The gate's or the +// assigner's key is not the teller's to take away: a later asking of it, under a new router id, is asked as any. +func (t *Teller) EndAsk(ask, why, by string) ends.Result { + t.mu.Lock() + defer t.mu.Unlock() + id, key := ask, "" + if s, rec, err := t.seenOf(ask); err != nil { + return ends.Result{Says: err.Error()} + } else if rec != nil { + key = s.Key + } else if g, grec, err := t.seenAt(gateRecordKey(ask)); err != nil { + return ends.Result{Says: err.Error()} + } else if grec != nil && g.ID != "" { + id, key = g.ID, ask + } else { + return ends.Result{Says: "the teller holds no ask under " + ask + ": name its router id, as operator-channel's asks show it, or the key of the gate's or the assigner's asks"} + } + r := ends.Result{ID: id, Key: key} + waited := false + if ws, rec, err := t.waiting(); err == nil && rec != nil { + if i := slices.IndexFunc(ws, func(x waitingAsk) bool { return x.ID == id }); i >= 0 { + if t.keepWaiting(slices.Delete(ws, i, i+1), rec) != nil { + r.Says = "the ask " + id + " waits for room, and the wait could not be written: it is not ended" + return r + } + waited = true + } + } + k, _, err := t.keptOf(id) + if err != nil { + r.Says = err.Error() + return r + } + if !open(k) { + r.Ended = waited + if waited { + r.Says = "the ask " + id + " waited for room at the router: it is dropped from the wait and asked no more" + } else { + r.Says = "the ask " + id + " is not open (answered, lapsed or ended already): it holds no place at the router, and nothing is done" + } + t.logf("%s: %s, by %s: %s", key, r.Says, by, why) + return r + } + t.cancelOpen(key, id) + if k, _, err := t.keptOf(id); err != nil || open(k) { + r.Says = "the ask " + id + " could not be cancelled at the router (see the teller's journal): it stays open" + return r + } + r.Ended = true + r.Says = "the ask " + id + " is cancelled at the router, which frees its place; the next ask waiting for room is asked when the router says it ended" + t.logf("%s: the ask %s is ended by hand, by %s: %s", key, id, by, why) + return r +} + +// endAsked answers one value of the state ends: one not yet answered is carried out and written back, by compare-and-set at +// its revision, with its result. One answered already (the teller's own write) is left. +func endAsked(ctx context.Context, t *Teller, kv jetstream.KeyValue, key string, rev uint64, value []byte) { + e, err := ends.Decode(value) + if err != nil { + t.logf("%s.%s holds what is not an end, left: %v", ends.State, key, err) + return + } + if e.Done() { + return + } + r := t.EndAsk(e.Ask, e.Why, e.By) + e.Result, e.Answered = &r, time.Now().UTC() + data, err := json.Marshal(e) + if err != nil { + return + } + if _, err := kv.Update(ctx, key, data, rev); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + t.logf("%s.%s: the answer could not be written: %v", ends.State, key, err) + } +} 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 4e2c9c8..ad4a99f 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -3,7 +3,8 @@ // 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. +// subject: its one tool, teller_end_ask (the program mesh-delivery-teller-tools, issue 774), asks through its state ends, +// which it watches. package main import ( @@ -22,6 +23,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/asker" "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-teller/internal/ends" ) func main() { @@ -74,7 +77,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // The view's answers to the assigner's asks (issue 1297 B10): both buckets are the controller's to create, and a bucket not // there is said and leaves the teller without that path rather than without the gate's asks (a wait for it needs a timer). - failed := make(chan error, 2) + failed := make(chan error, 4) assignerKV, viewKV := process.OptionalBucket(ctx, js, delivery.AssignerAsksBucket, teller, logger.Printf), process.OptionalBucket(ctx, js, delivery.ViewAnswersBucket, teller, logger.Printf) // The assigner's asks are raised on the operator channel whether or not the view's answers can be read (issue 1906). if viewKV != nil { @@ -91,6 +94,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro failed <- watchAsks(ctx, viewKV, func(key string, _ uint64, _ []byte) { t.OnViewAnswer(key) }, func(string) {}, nil) }() } + // A person ends one ask by hand with teller_end_ask, which writes the state ends (issue 774): watched as the gate's asks + // are, once it is on the bus. + process.AwaitBucket(ctx, js, conn, ends.Bucket, "teller_end_ask is not answered; the asks still end when they are answered, lapse or their key is taken away", logger.Printf).Ready(func(kv jetstream.KeyValue) { + go func() { + if err := watchAsks(ctx, kv, func(key string, rev uint64, value []byte) { endAsked(ctx, t, kv, key, rev, value) }, func(string) {}, nil); err != nil { + failed <- fmt.Errorf("watching the ends asked by hand: %w", err) + } + }() + }) go func() { if err := watchAsks(ctx, gate, t.OnAsk, t.OnAskGone, t.OnCurrent); err != nil { failed <- fmt.Errorf("watching the gate's asks: %w", err) 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 index af3e1c9..3f6aee0 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go @@ -287,6 +287,55 @@ func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1 if n := len(b.askedOf(asks.CancelSubject(delivery.TellerModule))); n != 1 { t.Errorf("the ask was cancelled %d times", n) } + // Issue 774 (2026-10-31): two merge asks of pull requests merged by another path held two of the router's three places, + // their keys never taken away, and a fourth ask was refused for want of room. teller_end_ask ends one by hand, named by its + // gate key or its router id; the router's word that it ended asks the one waiting for room again. + for _, c := range []struct { + name string + by func(key string, ga delivery.Ask) string + }{ + {"774: ended by hand, named by its gate key", func(key string, _ delivery.Ask) string { return key }}, + {"774: ended by hand, named by its router id", func(_ string, ga delivery.Ask) string { return ga.Ask.ID }}, + } { + t.Run(c.name, func(t *testing.T) { + b := newBed(t) + k1, first := b.askOf(t, 1, 1) + b.askOf(t, 2, 2) + b.askOf(t, 3, 3) + _, fourth := b.askOf(t, 4, 4) + b.refuse(t, fourth.Ask.ID, roomRefusal) + r := b.t.EndAsk(c.by(k1, first), "its pull request merged without the Approve", "lead@shanks") + cancels := b.askedOf(asks.CancelSubject(delivery.TellerModule)) + if !r.Ended || r.ID != first.Ask.ID || r.Key != k1 || len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+first.Ask.ID+`"}` { + t.Fatalf("end = %+v, cancels %+v", r, cancels) + } + b.deliver(t, asks.Warrant{Ask: first.Ask.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeCancelled, At: now, AskDigest: first.Ask.Digest()}) + if got := b.askedIDs(); len(got) != 5 || got[4] != fourth.Ask.ID { + t.Errorf("asks = %v, want the refused %s asked again once the ask ended by hand left", got, fourth.Ask.ID) + } + if again := b.t.EndAsk(k1, "again", "lead@shanks"); again.Ended || len(b.askedOf(asks.CancelSubject(delivery.TellerModule))) != 1 { + t.Errorf("a second end = %+v: an ended ask is left", again) + } + }) + } + t.Run("774: an ask waiting for room is dropped from the wait; a name the teller holds no ask under is said", func(t *testing.T) { + b := newBed(t) + _, first := b.askOf(t, 1, 1) + b.askOf(t, 2, 2) + b.askOf(t, 3, 3) + _, fourth := b.askOf(t, 4, 4) + b.refuse(t, fourth.Ask.ID, roomRefusal) + if r := b.t.EndAsk(fourth.Ask.ID, "asked about what no longer holds", "lead@shanks"); !r.Ended || len(b.askedOf(asks.CancelSubject(delivery.TellerModule))) != 0 { + t.Fatalf("end of a waiting ask = %+v", r) + } + b.deliver(t, warrantFor(first.Ask, "approve")) + if got := b.askedIDs(); len(got) != 4 { + t.Errorf("asks = %v: the ask ended while it waited was asked again", got) + } + if r := b.t.EndAsk("merge.nothing.here", "a typo", "lead@shanks"); r.Ended || !strings.Contains(r.Says, "holds no ask") { + t.Errorf("end of an unknown ask = %+v", r) + } + }) } func TestAWarrantThatHoldsWritesOneAnswerAndASecondWritesNone(t *testing.T) { // 1343/AC2 diff --git a/modules/mesh-delivery-teller/internal/ends/ends.go b/modules/mesh-delivery-teller/internal/ends/ends.go new file mode 100644 index 0000000..6fd80eb --- /dev/null +++ b/modules/mesh-delivery-teller/internal/ends/ends.go @@ -0,0 +1,79 @@ +// Package ends is the teller's recovery tool teller_end_ask and the value it writes (issue 774; the instruction "tools are +// how a module is spoken to"). An ask the teller holds open at the router keeps one of the three places the router gives +// its asker (asks.MostOpen) until it is answered, it lapses, or the gate or the assigner takes its key away. When that last +// never comes (on 2026-10-31 two merge asks of pull requests merged by another path held two places for a day, and a +// placement ask was refused for want of room), a person ends the ask by hand. The teller's process serves no request +// subject (ADR 0437 §4), so the tool writes an end into the teller's state ends, which the teller watches, carries out +// and answers there, as the indexer answers indexer_take_in. +package ends + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +const ( + Tool = "teller_end_ask" + // State is the teller's state of ends asked by hand: the bucket mesh-delivery-teller_ends. + State = "ends" + Bucket = delivery.TellerModule + "_" + State +) + +// Result is the teller's answer to an end: which ask it found (its router id and the gate's or the assigner's key it was +// asked for), whether this end ended it, and what it did in words. +type Result struct { + ID string `json:"id,omitempty"` + Key string `json:"key,omitempty"` + Ended bool `json:"ended"` + Says string `json:"says"` +} + +// End is a value of the state ends: the ask named (a router id, as operator-channel's asks show it, or the key of the +// gate's or the assigner's asks it was asked for), why, who and when, and, once the teller carried it out, its answer. +type End struct { + V int `json:"v"` + Ask string `json:"ask"` + Why string `json:"why"` + By string `json:"by,omitempty"` + At time.Time `json:"at"` + Result *Result `json:"result,omitempty"` + Answered time.Time `json:"answered,omitzero"` +} + +// Done says the end is answered. +func (e End) Done() bool { return !e.Answered.IsZero() && e.Result != nil } + +// Decode reads a value of the state ends. +func Decode(data []byte) (End, error) { + var e End + if err := json.Unmarshal(data, &e); err != nil { + return e, err + } + if e.V != 1 { + return e, fmt.Errorf("an ends value of version %d, not 1", e.V) + } + if e.Ask == "" || e.Why == "" { + return e, errors.New("an ends value naming no ask or no why") + } + return e, nil +} + +// New is the value teller_end_ask writes: its key and the value. The key is the router's spelling of the name given +// (delivery.AskID), which a bucket key takes whatever the name holds, and is the same for the same name, so a second end +// of one ask overwrites the first. +func New(ask, why, by string, at time.Time) (string, []byte, error) { + ask, why = strings.TrimSpace(ask), strings.TrimSpace(why) + if ask == "" { + return "", nil, errors.New("name the ask to end (ask): its router id, as operator-channel's asks show it, or the key of the gate's or the assigner's asks") + } + if why == "" { + return "", nil, errors.New("say why the ask is ended (why), in plain words: it is logged with who ended it") + } + data, err := json.Marshal(End{V: 1, Ask: ask, Why: why, By: by, At: at.UTC()}) + return delivery.AskID(ask), data, err +} diff --git a/modules/mesh-delivery-teller/manifest_test.go b/modules/mesh-delivery-teller/manifest_test.go index fde1855..2111e27 100644 --- a/modules/mesh-delivery-teller/manifest_test.go +++ b/modules/mesh-delivery-teller/manifest_test.go @@ -23,18 +23,20 @@ type state struct { } type artifact struct { - Name string `json:"name"` - Kind string `json:"kind"` - Language string `json:"language"` - System string `json:"system"` - From string `json:"from"` - Binary string `json:"binary"` + Name string `json:"name"` + Kind string `json:"kind"` + Language string `json:"language"` + System string `json:"system"` + From string `json:"from"` + Binary string `json:"binary"` + Loads []string `json:"loads"` } type manifest struct { Module string `json:"module"` Version string `json:"version"` Slug string `json:"slug"` + Tools []string `json:"tools"` Uses []string `json:"uses"` State []state `json:"state"` Reads []string `json:"reads"` @@ -77,8 +79,8 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 t.Errorf("module = %q", m.Module) } // 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) { + // gate key and revision each ask was made at (issue 1343); ends is the ends asked by hand with teller_end_ask (issue 774). + if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}, {Name: "ends", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } // The assigner's asks and the view's answers to them are read so the view's answer fills an ask that needs no proof (1297/B10). @@ -94,8 +96,12 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 if m.SecretsOwner != "mesh-delivery-teller" { t.Errorf("secrets-owner = %q", m.SecretsOwner) } - // No consumes, invokes, emits, streams, bus, seat or tool (ADR 0437 §4). - for _, k := range []string{"consumes", "invokes", "streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "tools", "seat", "seats", "bus"} { + // 774: teller_end_ask, served by its tools bundle through the state ends; the process serves none. + if want := []string{"teller_end_ask"}; !reflect.DeepEqual(m.Tools, want) { + t.Errorf("tools = %v, want %v", m.Tools, want) + } + // No consumes, invokes, emits, streams, bus or seat (ADR 0437 §4). + for _, k := range []string{"consumes", "invokes", "streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the teller has none", k) } @@ -127,13 +133,18 @@ func TestManifestProcess(t *testing.T) { // 1240/AC1 if _, bus := p["bus"]; bus { t.Errorf("the process declares a bus: it runs on its own broker credential") } - if len(m.Build.Artifacts) != 1 { - t.Fatalf("artifacts = %+v", m.Build.Artifacts) + if len(m.Build.Artifacts) != 2 { + t.Fatalf("artifacts = %+v, want the program and the tools", m.Build.Artifacts) } a := m.Build.Artifacts[0] if st, err := os.Stat(a.From); err != nil || !st.IsDir() || a.Binary != "mesh-delivery-teller" || !strings.HasSuffix(a.From, a.Binary) { t.Errorf("artifact = %+v", a) } + if a := m.Build.Artifacts[1]; a.Name != "tools" || a.Kind != "bundle" || a.Binary != "mesh-delivery-teller-tools" || !reflect.DeepEqual(a.Loads, []string{"mesh-delivery-teller-tools"}) { + t.Errorf("tools artifact = %+v", a) // 774 + } else if st, err := os.Stat(a.From); err != nil || !st.IsDir() { + t.Errorf("the tools' source %s: %v", a.From, err) + } } func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1240/AC3 (ADR 0437 §8) diff --git a/modules/mesh-delivery-teller/module.json b/modules/mesh-delivery-teller/module.json index 2ec994e..6d814c0 100644 --- a/modules/mesh-delivery-teller/module.json +++ b/modules/mesh-delivery-teller/module.json @@ -2,9 +2,13 @@ "module": "mesh-delivery-teller", "version": "1", "slug": "teller", + "tools": [ + "teller_end_ask" + ], "state": [ { "name": "answers", "history": 5 }, - { "name": "asked", "history": 1 } + { "name": "asked", "history": 1 }, + { "name": "ends", "history": 1 } ], "reads": [ "mesh-delivery-assigner.asks", @@ -60,6 +64,17 @@ "system": "arch", "from": "cmd/mesh-delivery-teller", "binary": "mesh-delivery-teller" + }, + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/mesh-delivery-teller-tools", + "binary": "mesh-delivery-teller-tools", + "loads": [ + "mesh-delivery-teller-tools" + ] } ] }