List holds and let the operator release or keep one again from mesh-delivery-cli (issue 1906)

A hold answered keep on the phone could not be changed: its ask ended with the
answer, nothing listed it, and a kept hold is never lifted by itself. The view
lists every hold with its node, cause and answer (mesh-delivery-cli holds) and
answers the standing hold ask with hold-release or hold-keep through its one
write; the teller takes a later view answer that changes a hold's option into
the answer of that asking, never an older one.

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go

@@ -641,3 +641,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())
+}

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go

@@ -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
+}

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go

@@ -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
+}

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go

@@ -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"
@@ -175,3 +176,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")
+	}
+}