diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go index 99a2895..bb24aa4 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go @@ -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()) +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go index 288d07d..13f7ecd 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go +++ b/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 +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go index f41b914..7f64f52 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go +++ b/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 +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go index d6cea8f..d53985e 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go +++ b/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" @@ -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") + } +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main.go index d5c1e29..24a732b 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main.go @@ -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