Merge pull request 'List holds and let the operator release or keep one again from mesh-delivery-cli (issue 1906)' (#41) from fix/r6-hold-verbs into main

This commit was merged in pull request #41.
This commit is contained in:
2026-10-25 14:30:12 +00:00
10 changed files with 593 additions and 14 deletions
@@ -524,8 +524,9 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
if a, err := delivery.DecodeAnswer(e.Value()); err != nil || a.Option != "release" || answeredAsking(e.Value()) != ask.ID {
t.Fatalf("answer %+v err %v", a, err)
}
// The view's answer written again changes nothing under the ask.
if _, err := viewKV.Put(b.ctx, key+".view", viewAnswer(t, "keep", "")); err != nil {
// The view's answer written again with the same option changes nothing under the ask (another option, later, is the
// operator changing their mind: TestTellerTakesAChangedHoldAnswerFromTheView, issue 1906 note 23).
if _, err := viewKV.Put(b.ctx, key+".view", viewAnswer(t, "release", "")); err != nil {
t.Fatal(err)
}
time.Sleep(time.Second)
@@ -641,3 +642,52 @@ func TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer(t *testing.T) {
t.Errorf("the ask taken away was not cancelled: %v %v\n%s", raw, err, logs.String())
}
}
// TestTellerTakesAChangedHoldAnswerFromTheView starts the program against the bus: a hold ask answered "keep" earlier, and the
// view's answer "release" written now, replace the answer of that asking with "release" (1906/N23 on the real bus).
func TestTellerTakesAChangedHoldAnswerFromTheView(t *testing.T) {
now = time.Now().UTC()
b := startFull(t)
assigner, view := b.as(t, "n1.mesh-delivery-assigner"), b.as(t, "n1.mesh-delivery-view")
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)
}
key, value, ask := holdAsk(t)
answers, err := b.js.KeyValue(b.ctx, tellerAnswers)
if err != nil {
t.Fatal(err)
}
kept, _ := delivery.EncodeAnswer(delivery.Answer{Option: "keep", Warrant: ask.ID + "@" + time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano), Words: "the operator chose keep"})
if _, err := answers.Put(b.ctx, delivery.AnswerKey(key), kept); err != nil {
t.Fatal(err)
}
askKV, err := assigner.KeyValue(b.ctx, assignerAsks)
if err != nil {
t.Fatal(err)
}
if _, err := askKV.Put(b.ctx, key, value); 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) })
viewKV, err := view.KeyValue(b.ctx, viewAnswers)
if err != nil {
t.Fatal(err)
}
time.Sleep(500 * time.Millisecond) // the program is watching
if _, err := viewKV.Put(b.ctx, key+".view", viewAnswer(t, "release", "it was a flake")); err != nil {
t.Fatal(err)
}
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 {
if a, err := delivery.DecodeAnswer(e.Value()); err == nil && a.Option == "release" && answeredAsking(e.Value()) == ask.ID {
return
}
}
}
t.Fatalf("the kept hold's answer was not changed to release\n%s", logs.String())
}
@@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"fmt"
"time"
"github.com/nats-io/nats.go/jetstream"
@@ -107,3 +108,15 @@ func (g kvGate) Get(key string) ([]byte, bool, error) {
}
return e.Value(), true, nil
}
// GetAt is Get with the time the value was written.
func (g kvGate) GetAt(key string) ([]byte, time.Time, bool, error) {
e, err := g.kv.Get(g.ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, time.Time{}, false, nil
}
if err != nil {
return nil, time.Time{}, false, err
}
return e.Value(), e.Created(), true, nil
}
@@ -2,6 +2,7 @@ package main
import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -39,7 +40,10 @@ func (t *Teller) OnAssignerAsk(key string) {
}
// fillFromView writes the answer to the assigner's ask at askKey from the view's answer, when both are there and readable, the
// ask is of a kind that needs no proof, the option is one the ask offers and the asking is not answered yet.
// ask is of a kind that needs no proof, the option is one the ask offers and the asking is not answered yet. A hold ask answered
// already takes a later answer of the view that chose another option: the operator changed their mind (issue 1906 note 23: a
// "keep" given on the phone could not be undone). Later means the view's answer was written after the answer that stands, so a
// view answer read again at a start never replaces a newer answer from the phone.
func (t *Teller) fillFromView(askKey string) {
if t.Assigner == nil || t.View == nil {
return
@@ -49,7 +53,7 @@ func (t *Teller) fillFromView(askKey string) {
t.logf("%s is not answered from the view: %v", askKey, err)
return
}
rawAnswer, found, err := t.View.Get(viewKey)
rawAnswer, at, found, err := t.viewAnswer(viewKey)
if err != nil {
t.logf("%s: the view's answer could not be read, so the ask is left: %v", askKey, err)
return
@@ -85,14 +89,14 @@ func (t *Teller) fillFromView(askKey string) {
t.logf("%s: the answer could not be read: %v", askKey, err)
return
}
if has && answeredAsking(cur) == ask.Ask.ID {
if has && answeredAsking(cur) == ask.Ask.ID && !t.changesMind(askKey, cur, va, at) {
return // this asking is answered, from the view or elsewhere
}
words := "answered in the view (mesh-delivery-view): " + va.Option
if w := strings.TrimSpace(va.Words); w != "" {
words += ": " + w
}
body, err := delivery.EncodeAnswer(delivery.Answer{Option: va.Option, Warrant: ask.Ask.ID + "@" + t.now().UTC().Format("2006-01-02T15:04:05.999999999Z07:00"), Words: words})
body, err := delivery.EncodeAnswer(delivery.Answer{Option: va.Option, Warrant: ask.Ask.ID + "@" + at.UTC().Format(time.RFC3339Nano), Words: words})
if err != nil {
t.logf("%s: the answer could not be made: %v", askKey, err)
return
@@ -108,3 +112,45 @@ func (t *Teller) fillFromView(askKey string) {
t.cancelOpen(askKey, ask.Ask.ID) // answered in the view: the same ask on the phone is taken back
}
}
// timedAnswers is the view's bucket as the program reads it: each answer with the time it was written.
type timedAnswers interface {
GetAt(key string) (value []byte, at time.Time, found bool, err error)
}
// viewAnswer is the view's answer at key and when it was written; a bucket that keeps no time gives the teller's now.
func (t *Teller) viewAnswer(key string) ([]byte, time.Time, bool, error) {
if tv, ok := t.View.(timedAnswers); ok {
v, at, found, err := tv.GetAt(key)
if err == nil && !at.IsZero() {
return v, at, found, nil
}
if err != nil {
return nil, time.Time{}, false, err
}
return v, t.now(), found, nil
}
v, found, err := t.View.Get(key)
return v, t.now(), found, err
}
// changesMind says whether the view's answer va, written at at, replaces the answer cur of the same asking of the hold ask at
// askKey: it chose another option and was written after cur's warrant.
func (t *Teller) changesMind(askKey string, cur []byte, va delivery.ViewAnswer, at time.Time) bool {
k, err := delivery.ParseAnyAskKey(askKey)
if err != nil || k.Kind != delivery.AskKindHold {
return false
}
a, err := delivery.DecodeAnswer(cur)
if err != nil || a.Option == va.Option {
return false
}
_, when, _ := strings.Cut(a.Warrant, "@")
before, err := time.Parse(time.RFC3339Nano, when)
if err != nil || !at.After(before) {
t.logf("%s: the view's answer %s is not newer than the answer %s that stands (%s), and is not taken", askKey, va.Option, a.Option, when)
return false
}
t.logf("%s: the operator changed the answer from %s to %s in the view", askKey, a.Option, va.Option)
return true
}
@@ -3,6 +3,7 @@ package main
import (
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/asks"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -118,8 +119,11 @@ func TestAViewAnswerOfAnOptionTheAskDoesNotOfferFillsNothing(t *testing.T) { //
}
}
// A view answer written before the answer that stands never replaces it (1906/N23: only a later one may, for a hold). The
// view's time is fixed here, not the package's clock, which the integration tests set to the wall clock.
func TestAViewAnswerNeverReplacesTheAnswerOfTheSameAsking(t *testing.T) { // 1297/B10
b := newBed(t)
b.t.Now = func() time.Time { return time.Date(2026, 10, 19, 12, 0, 0, 0, time.UTC) }
assigner, view := withView(b)
key, value, ask := holdAsk(t)
assigner[key] = value
@@ -175,3 +179,53 @@ func TestAStaleViewAnswerOfAnotherAskingFillsNothing(t *testing.T) { // 1297/B10
t.Errorf("writes %d; log:\n%s", b.answers.writes, b.logs.String())
}
}
// timedView is the view's bucket with the time each answer was written.
type timedView struct {
memGate
at map[string]time.Time
}
func (v timedView) GetAt(key string) ([]byte, time.Time, bool, error) {
b, ok := v.memGate[key]
return b, v.at[key], ok, nil
}
// 1906/N23-AC4: given a hold ask answered "keep" on the phone, when the operator answers "release" in the view later, then the
// teller replaces the answer of that asking with "release", whose warrant names the asking and the view's time; a view answer
// written before the phone's answer, the same option, or an ask of another kind replaces nothing.
func TestALaterViewAnswerChangesAHoldsAnswer(t *testing.T) {
setup := func(viewAt time.Time, option string) (*bed, string, asks.Ask) {
b := newBed(t)
assigner := memGate{}
view := timedView{memGate: memGate{}, at: map[string]time.Time{}}
b.t.Assigner, b.t.View = assigner, view
key, value, ask := holdAsk(t)
assigner[key] = value
kept, _ := delivery.EncodeAnswer(delivery.Answer{Option: "keep", Warrant: ask.ID + "@" + now.Add(time.Hour).UTC().Format(time.RFC3339Nano), Words: "the operator, via telegram, chose keep"})
b.answers.written[key], b.answers.revs[key] = kept, 3
view.memGate[key+".view"] = viewAnswer(t, option, "it was a flake")
view.at[key+".view"] = viewAt
b.t.OnViewAnswer(key + ".view")
return b, key, ask
}
later := now.Add(2 * time.Hour)
b, key, ask := setup(later, "release")
a, err := delivery.DecodeAnswer(b.answers.written[key])
if err != nil || b.answers.writes != 1 || a.Option != "release" || a.Warrant != ask.ID+"@"+later.UTC().Format(time.RFC3339Nano) {
t.Fatalf("writes %d, answer %+v, %v; log:\n%s", b.answers.writes, a, err, b.logs.String())
}
for name, b := range map[string]*bed{
"older": func() *bed { b, _, _ := setup(now, "release"); return b }(),
"same": func() *bed { b, _, _ := setup(later, "keep"); return b }(),
} {
if b.answers.writes != 0 {
t.Errorf("%s: the answer was replaced", name)
}
}
// A restore ask answered stays answered: only a hold's answer may be changed.
bb := newBed(t)
if bb.t.changesMind("restore.ace.plex.2222222222222222222222222222222222222222222222222222222222222222", b.answers.written[key], delivery.ViewAnswer{Option: "keep"}, later.Add(time.Hour)) {
t.Error("a restore ask's answer was changed")
}
}
@@ -1,6 +1,8 @@
// Command mesh-delivery-cli is the operator's command for the delivery view (novox/hq ADR 0429 §3). It asks the
// view's local socket for the rows and prints them; it holds no credential. `answer <ask key> <option> [words]` asks the view
// to write the operator's answer to an ask that needs no proof of who answered (the view refuses the merge's Approve).
// to write the operator's answer to an ask that needs no proof of who answered (the view refuses the merge's Approve). `holds`
// lists every fingerprint the assigner holds with its node, cause and answer; `hold-release <module> <fingerprint> [words]` and
// `hold-keep <module> <fingerprint> [words]` answer its hold ask, again when the operator changes their mind (issue 1906).
package main
import (
@@ -10,6 +12,7 @@ import (
"io"
"os"
"strings"
"text/tabwriter"
"mesh-delivery-view/internal/view"
)
@@ -31,9 +34,26 @@ func run(args []string, socket string, out, errOut io.Writer) int {
if err := fs.Parse(args); err != nil {
return 2
}
if fs.NArg() > 0 && fs.Arg(0) == "holds" {
return holds(socket, *asJSON, out, errOut)
}
if fs.NArg() > 0 && (fs.Arg(0) == "hold-release" || fs.Arg(0) == "hold-keep") {
if fs.NArg() < 3 {
fmt.Fprintf(errOut, "mesh-delivery-cli %s <module> <fingerprint> [words]\n", fs.Arg(0))
return 2
}
option := strings.TrimPrefix(fs.Arg(0), "hold-")
r, err := view.AnswerHoldOn(socket, fs.Arg(1), fs.Arg(2), option, strings.Join(fs.Args()[3:], " "))
if err != nil {
fmt.Fprintln(errOut, "mesh-delivery-cli:", err)
return 1
}
fmt.Fprintf(out, "answered the hold of %s at %s with %s; the teller fills the ask from it and the assigner reads it\n", r.Module, r.FP, option)
return 0
}
if fs.NArg() > 0 {
if fs.Arg(0) != "answer" || fs.NArg() < 3 {
fmt.Fprintln(errOut, "mesh-delivery-cli shows the rows, or answers: mesh-delivery-cli answer <ask key> <option> [words]")
fmt.Fprintln(errOut, "mesh-delivery-cli shows the rows, or answers: mesh-delivery-cli answer <ask key> <option> [words]; holds; hold-release|hold-keep <module> <fingerprint> [words]")
return 2
}
key, option := fs.Arg(1), fs.Arg(2)
@@ -61,3 +81,53 @@ func run(args []string, socket string, out, errOut io.Writer) int {
view.Table(out, s)
return 0
}
// holds prints every hold: module, fingerprint, node, since, answer, words.
func holds(socket string, asJSON bool, out, errOut io.Writer) int {
rows, err := view.AskHolds(socket)
if err != nil {
fmt.Fprintln(errOut, "mesh-delivery-cli:", err)
return 1
}
if asJSON {
enc := json.NewEncoder(out)
enc.SetIndent("", " ")
if err := enc.Encode(rows); err != nil {
fmt.Fprintln(errOut, "mesh-delivery-cli:", err)
return 1
}
return 0
}
if len(rows) == 0 {
fmt.Fprintln(out, "the assigner holds nothing")
return 0
}
tw := tabwriter.NewWriter(out, 0, 2, 2, ' ', 0)
fmt.Fprintln(tw, "MODULE\tFINGERPRINT\tNODE\tSINCE\tANSWER\tWORDS")
for _, r := range rows {
answer := r.Answer
switch {
case r.AskKey == "":
answer = "no ask"
case r.Pending != "":
answer = r.Pending + " (from the view, the teller fills it)"
case answer == "":
answer = "not answered"
}
fp := strings.TrimPrefix(r.FP, "src2:")
if len(fp) > 12 {
fp = fp[:12]
}
since := ""
if !r.Since.IsZero() {
since = r.Since.Local().Format("2006-01-02 15:04")
}
words := r.Words
if r.Error != "" {
words = "unreadable: " + r.Error
}
fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s\n", r.Module, fp, r.Node, since, answer, words)
}
tw.Flush()
return 0
}
@@ -240,3 +240,46 @@ func TestTheAnswerCommandAnswersAnAskNeedingNoProof(t *testing.T) { // 1297/B9b
}
}
}
// 1906/N23-AC3: `holds` lists the holds with node, cause and answer; `hold-release <module> <fingerprint>` and `hold-keep` answer
// the hold ask through the view, also one answered already.
func TestTheHoldCommandsListAndAnswerHolds(t *testing.T) {
fp := "src2:04efd1a1aa" + strings.Repeat("2", 54)
key, err := delivery.HoldAskKey("systemd-resolved", fp)
if err != nil {
t.Fatal(err)
}
hold := asks.Ask{ID: delivery.AskID(key), Headline: "systemd-resolved is held", Explanation: "ace failed the check.", Who: asks.Operator, Expires: time.Now().Add(time.Hour), OnExpiry: "stays held",
Options: []asks.Option{{ID: "keep", Label: "Keep", Does: "stays", Level: asks.Acknowledge}, {ID: "release", Label: "Release", Does: "goes on", Level: asks.Acknowledge}}}
value, _ := delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: hold})
hkey, _ := delivery.HeldKey("systemd-resolved", fp)
held, _ := delivery.EncodeHeld(delivery.Held{Node: "ace", Words: "health failed, phase running: dns timeout"})
ans, _ := delivery.EncodeAnswer(delivery.Answer{Option: "keep", Warrant: hold.ID + "@2026-10-25T04:00:00Z", Words: "the operator chose keep"})
m, w := view.New(), &putter{puts: map[string][]byte{}}
m.SetWriter(w)
m.Apply(view.Entry{Bucket: delivery.AssignerAsksBucket, Key: key, Value: value})
m.Apply(view.Entry{Bucket: delivery.AssignerWantsBucket, Key: hkey, Value: held})
m.Apply(view.Entry{Bucket: delivery.AnswersBucket, Key: key, Value: ans})
socket := serveWith(t, m)
var out, errOut bytes.Buffer
if code := run([]string{"holds"}, socket, &out, &errOut); code != 0 {
t.Fatalf("exit %d: %s", code, errOut.String())
}
for _, want := range []string{"systemd-resolved", "04efd1a1aa22", "ace", "keep", "dns timeout"} {
if !strings.Contains(out.String(), want) {
t.Errorf("holds printed %q, missing %q", out.String(), want)
}
}
out.Reset()
if code := run([]string{"hold-release", "systemd-resolved", "04efd1a1", "a", "flake"}, socket, &out, &errOut); code != 0 {
t.Fatalf("exit %d: %s", code, errOut.String())
}
if a, err := delivery.DecodeViewAnswer(w.puts[key+".view"]); err != nil || a.Option != "release" || a.Asking != hold.ID || a.Words != "a flake" {
t.Errorf("the view's answer = %+v, %v", a, err)
}
errOut.Reset()
if code := run([]string{"hold-keep", "systemd-resolved"}, socket, &out, &errOut); code != 2 {
t.Errorf("hold-keep without a fingerprint: exit %d", code)
}
}
@@ -0,0 +1,188 @@
package view
import (
"encoding/json"
"errors"
"fmt"
"net"
"slices"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/asks"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// The assigner's holds, listed and answered from the operator's command (issue 1906 note 23). A hold ask answered once (on the
// phone or in the view) ended with that answer, so nothing listed it any more and an answer "keep" could not be changed: the
// operator kept systemd-resolved held and then could not release it, and a kept hold is never lifted by itself. Now the view
// lists every hold with its node, its cause and its answer, and answers the hold ask that stands, again if the operator changes
// their mind, through the view's one write (`<ask key>.view`), which the teller takes into the ask's answer. Nothing else writes:
// the answer still reaches the assigner only through the teller (mesh-delivery-teller viewanswers.go).
// The options of the hold ask (mesh-delivery-assigner internal/assign asks.go HoldAsk).
const (
HoldRelease = "release"
HoldKeep = "keep"
)
type heldEntry struct {
module, fp string
value delivery.Held
err string
at time.Time
}
// applyHeld takes a change of a held key of the assigner's wants, and says whether the key was one.
func (m *Model) applyHeld(e Entry) bool {
module, fp, err := delivery.ParseHeldKey(e.Key)
if err != nil {
return false
}
if e.Deleted {
delete(m.holds, e.Key)
return true
}
h, err := delivery.DecodeHeld(e.Value)
m.holds[e.Key] = heldEntry{module: module, fp: fp, value: h, err: errText(err), at: e.At}
return true
}
// HoldRow is one fingerprint the assigner holds: the module, the fingerprint, the node whose health check failed, the hold's
// words (its cause) and since when; the ask about it (its key and the asking's id, empty when the assigner asks none), and the
// answer that stands for that asking (the option and its words, empty when none), with the view's own answer when it waits for
// the teller.
type HoldRow struct {
Module string `json:"module"`
FP string `json:"fp"`
Node string `json:"node,omitempty"`
Words string `json:"words,omitempty"`
Since time.Time `json:"since,omitempty"`
AskKey string `json:"ask-key,omitempty"`
Asking string `json:"asking,omitempty"`
Answer string `json:"answer,omitempty"`
Answered string `json:"answered,omitempty"`
Pending string `json:"pending,omitempty"`
Error string `json:"error,omitempty"`
}
// Holds is every hold, ordered by module and fingerprint.
func (m *Model) Holds() []HoldRow {
m.mu.Lock()
defer m.mu.Unlock()
out := []HoldRow{}
for _, h := range m.holds {
row := HoldRow{Module: h.module, FP: h.fp, Node: h.value.Node, Words: h.value.Words, Since: h.at, Error: h.err}
if key, err := delivery.HoldAskKey(h.module, h.fp); err == nil {
if a, ok := m.aasks[key]; ok && a.err == "" {
row.AskKey, row.Asking = key, a.ask.ID
if ans, ok := m.answerOf[key]; ok && m.answers[key] == a.ask.ID {
row.Answer, row.Answered = ans.Option, ans.Words
}
if c := m.chosenOf(key, a.ask); c != "" && c != row.Answer {
row.Pending = c
}
}
}
out = append(out, row)
}
sort.Slice(out, func(i, j int) bool {
if out[i].Module != out[j].Module {
return out[i].Module < out[j].Module
}
return out[i].FP < out[j].FP
})
return out
}
// AnswerHold writes the view's answer to the hold ask of module at the fingerprint fp: release or keep, with the operator's
// words. fp is the whole fingerprint, or the start of its hex (as the rows print it), naming one hold of the module. Unlike
// Answer it answers an ask answered already, so the operator can change their mind; an answer equal to the one that stands is
// refused, as is a hold the assigner asks nothing about.
func (m *Model) AnswerHold(module, fp, option, words string) (HoldRow, error) {
option = strings.TrimSpace(option)
if option != HoldRelease && option != HoldKeep {
return HoldRow{}, fmt.Errorf("a hold is answered %s or %s, not %q", HoldRelease, HoldKeep, option)
}
var match []HoldRow
want := strings.TrimPrefix(strings.TrimSpace(fp), delivery.SourcePrefix2)
for _, r := range m.Holds() {
if r.Module == module && want != "" && strings.HasPrefix(strings.TrimPrefix(r.FP, delivery.SourcePrefix2), want) {
match = append(match, r)
}
}
switch {
case len(match) == 0:
return HoldRow{}, fmt.Errorf("the assigner holds no fingerprint %q of %s (mesh-delivery-cli holds lists them)", fp, module)
case len(match) > 1:
return HoldRow{}, fmt.Errorf("%q names %d holds of %s: give more of the fingerprint", fp, len(match), module)
}
r := match[0]
if r.AskKey == "" {
return r, fmt.Errorf("the assigner asks nothing about the hold of %s at %s, so there is nothing to answer", module, r.FP)
}
if r.Answer == option && r.Pending == "" {
return r, fmt.Errorf("the hold of %s at %s is answered %s already", module, r.FP, option)
}
m.mu.Lock()
a, w := m.aasks[r.AskKey], m.writer
m.mu.Unlock()
if !slices.ContainsFunc(a.ask.Options, func(o asks.Option) bool { return o.ID == option }) {
return r, fmt.Errorf("the hold ask %s offers no option %q", r.AskKey, option)
}
if w == nil {
return r, fmt.Errorf("the view cannot write its answers here")
}
key, err := delivery.ViewAnswerKey(r.AskKey)
if err != nil {
return r, err
}
body, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Asking: a.ask.ID, Option: option, Words: strings.TrimSpace(words)})
if err != nil {
return r, err
}
if err := w.Put(key, body); err != nil {
return r, err
}
return r, nil
}
// askOn sends one ask to the socket at path and decodes the reply into out; the view's refusal is the error.
func askOn(path string, req ask, out any) error {
c, err := net.DialTimeout("unix", path, 5*time.Second)
if err != nil {
return fmt.Errorf("cannot reach the view at %s: %w", path, err)
}
defer c.Close()
c.SetDeadline(time.Now().Add(10 * time.Second))
raw, _ := json.Marshal(req)
if _, err := c.Write(append(raw, '\n')); err != nil {
return err
}
var reply json.RawMessage
if err := json.NewDecoder(c).Decode(&reply); err != nil {
return fmt.Errorf("the view's reply cannot be read: %w", err)
}
var refusal struct {
Error string `json:"error"`
}
if json.Unmarshal(reply, &refusal) == nil && refusal.Error != "" {
return errors.New(refusal.Error)
}
return json.Unmarshal(reply, out)
}
// AskHolds asks the socket at path for every hold.
func AskHolds(path string) ([]HoldRow, error) {
var out []HoldRow
err := askOn(path, ask{Ask: "holds"}, &out)
return out, err
}
// AnswerHoldOn asks the socket at path to answer the hold ask of module at fp with option.
func AnswerHoldOn(path, module, fp, option, words string) (HoldRow, error) {
var out HoldRow
err := askOn(path, ask{Ask: "hold", Module: module, FP: fp, Option: option, Words: words}, &out)
return out, err
}
@@ -0,0 +1,91 @@
package view
import (
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// heldE is the assigner's hold of module at fp, as it writes it in its wants.
func heldE(t *testing.T, module, fp, node, words string) Entry {
t.Helper()
key, err := delivery.HeldKey(module, fp)
if err != nil {
t.Fatal(err)
}
v, err := delivery.EncodeHeld(delivery.Held{Node: node, Words: words})
if err != nil {
t.Fatal(err)
}
return Entry{Bucket: delivery.AssignerWantsBucket, Key: key, Value: v, At: t0}
}
func answered(t *testing.T, key, id, option string) Entry {
t.Helper()
v, err := delivery.EncodeAnswer(delivery.Answer{Option: option, Warrant: id + "@2026-10-25T04:00:00Z", Words: "the operator, via telegram, chose " + option})
if err != nil {
t.Fatal(err)
}
return Entry{Bucket: delivery.AnswersBucket, Key: key, Value: v, At: t0}
}
// 1906/N23-AC1: given holds in the assigner's wants, the view lists each with its node, its cause in words, its hold ask and
// the answer that stands for that asking; a hold without an ask says so, and a held key is never taken for a want.
func TestTheViewListsEveryHoldWithItsNodeCauseAndAnswer(t *testing.T) {
m := New()
e, a := holdAsk(t)
fill(m, e, heldE(t, "plex", fpNew, "ace", "health failed, phase running: connection refused"), answered(t, e.Key, a.ID, "keep"),
heldE(t, "web", fpNew, "g14", "health failed"))
got := m.Holds()
if len(got) != 2 {
t.Fatalf("holds = %+v", got)
}
p := got[0]
if p.Module != "plex" || p.FP != fpNew || p.Node != "ace" || !strings.Contains(p.Words, "connection refused") || p.AskKey != e.Key || p.Asking != a.ID || p.Answer != "keep" {
t.Errorf("plex = %+v", p)
}
if w := got[1]; w.Module != "web" || w.Node != "g14" || w.AskKey != "" || w.Answer != "" {
t.Errorf("web = %+v", w)
}
if len(m.Snapshot().Wants) != 0 {
t.Errorf("a hold was taken for a want: %+v", m.Snapshot().Wants)
}
}
// 1906/N23-AC2: given a hold ask answered "keep" (the ask ended with it), when the operator releases the hold, then the view
// writes its answer "release" to that asking under the ask's view key, though the ask was answered; the fingerprint may be the
// start of its hex. The same answer again, an unknown hold and another option are refused, and write nothing.
func TestAKeptHoldCanBeReleased(t *testing.T) {
m, w := New(), &memWriter{}
m.SetWriter(w)
e, a := holdAsk(t)
fill(m, e, heldE(t, "plex", fpNew, "ace", "health failed"), answered(t, e.Key, a.ID, "keep"))
short := strings.TrimPrefix(fpNew, delivery.SourcePrefix2)[:8]
if _, err := m.AnswerHold("plex", short, HoldRelease, "it was a flake"); err != nil {
t.Fatal(err)
}
va, err := delivery.DecodeViewAnswer(w.puts[e.Key+".view"])
if err != nil || va.Asking != a.ID || va.Option != HoldRelease || va.Words != "it was a flake" {
t.Fatalf("the view's answer = %+v, %v", va, err)
}
w.puts = nil
fill(m, answered(t, e.Key, a.ID, HoldRelease))
for name, try := range map[string]func() error{
"again": func() error { _, err := m.AnswerHold("plex", fpNew, HoldRelease, ""); return err },
"unknown": func() error { _, err := m.AnswerHold("plex", "ffff", HoldKeep, ""); return err },
"option": func() error { _, err := m.AnswerHold("plex", fpNew, "restore", ""); return err },
"module": func() error { _, err := m.AnswerHold("web", fpNew, HoldKeep, ""); return err },
} {
if err := try(); err == nil {
t.Errorf("%s: not refused", name)
}
}
if len(w.puts) != 0 {
t.Errorf("a refused answer wrote %v", w.puts)
}
// And kept again: the operator may change their mind both ways.
if _, err := m.AnswerHold("plex", fpNew, HoldKeep, ""); err != nil {
t.Errorf("keep after release: %v", err)
}
}
@@ -46,6 +46,8 @@ type ask struct {
Key string `json:"key,omitempty"`
Option string `json:"option,omitempty"`
Words string `json:"words,omitempty"`
Module string `json:"module,omitempty"`
FP string `json:"fp,omitempty"`
}
// ServeSocket answers {"ask":"rows"} on l, line-delimited JSON, until ctx ends, and {"ask":"answer","key":…,"option":…,"words":…}
@@ -72,8 +74,21 @@ func answer(c net.Conn, m *Model) {
return
}
var a ask
if json.Unmarshal(line, &a) != nil || (a.Ask != "rows" && a.Ask != "answer") {
json.NewEncoder(c).Encode(map[string]string{"error": `refused: this socket answers {"ask":"rows"} and {"ask":"answer"} and nothing else`})
if json.Unmarshal(line, &a) != nil || (a.Ask != "rows" && a.Ask != "answer" && a.Ask != "holds" && a.Ask != "hold") {
json.NewEncoder(c).Encode(map[string]string{"error": `refused: this socket answers {"ask":"rows"}, {"ask":"answer"}, {"ask":"holds"} and {"ask":"hold"} and nothing else`})
return
}
switch a.Ask {
case "holds":
json.NewEncoder(c).Encode(m.Holds())
return
case "hold":
r, err := m.AnswerHold(a.Module, a.FP, a.Option, a.Words)
if err != nil {
json.NewEncoder(c).Encode(map[string]string{"error": err.Error()})
return
}
json.NewEncoder(c).Encode(r)
return
}
if a.Ask == "answer" {
@@ -231,8 +231,10 @@ type Model struct {
chosen map[string]delivery.ViewAnswer // the answers the view wrote, by ask key
writer Writer
backlog *Backlog
answers map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable)
newest map[string]string // module -> id
answers map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable)
answerOf map[string]delivery.Answer // by key: the teller's answer whole, when readable (holds.go)
holds map[string]heldEntry // the assigner's holds, by held key (holds.go)
newest map[string]string // module -> id
haves map[string]haveEntry
nodes map[string]nodeEntry
// the watchman's buckets, by key
@@ -344,7 +346,7 @@ type nodeEntry struct {
// New is an empty model.
func New() *Model {
return &Model{assigns: map[string]assignment{}, builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, asks: map[string]askRecord{}, wants: map[string]wantEntry{}, told: map[string]bool{}, initial: map[string]bool{}, composed: map[string]composedEntry{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]delivery.ViewAnswer{}, answers: map[string]string{}, newest: map[string]string{},
return &Model{assigns: map[string]assignment{}, builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, asks: map[string]askRecord{}, wants: map[string]wantEntry{}, told: map[string]bool{}, initial: map[string]bool{}, composed: map[string]composedEntry{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]delivery.ViewAnswer{}, answers: map[string]string{}, answerOf: map[string]delivery.Answer{}, holds: map[string]heldEntry{}, newest: map[string]string{},
haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{},
presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}}
}
@@ -486,9 +488,12 @@ func (m *Model) apply(e Entry) {
m.chosen[askKey] = a
}
case delivery.AssignerWantsBucket:
if m.applyHeld(e) {
return
}
node, module, err := delivery.ParseWantKey(e.Key)
if err != nil {
return // a hold, and the node's own key, are not a module's want
return // the node's own key is not a module's want
}
key, kerr := delivery.HaveKey(node, module)
if kerr != nil {
@@ -520,6 +525,7 @@ func (m *Model) apply(e Entry) {
case delivery.AnswersBucket:
if e.Deleted {
delete(m.answers, e.Key)
delete(m.answerOf, e.Key)
return
}
// The answer's warrant is `<router id>@<time>`: it ends that asking, and only that one (a later asking of the key is
@@ -527,6 +533,9 @@ func (m *Model) apply(e Entry) {
id := ""
if a, err := delivery.DecodeAnswer(e.Value); err == nil {
id, _, _ = strings.Cut(a.Warrant, "@")
m.answerOf[e.Key] = a
} else {
delete(m.answerOf, e.Key)
}
m.answers[e.Key] = id
case delivery.BuilderBuildsBucket: