novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Take the assigner's ask in the SDK's layout and name the asking in the view's answer, so a stale answer never fills a later asking (issue 1297 B9 and B10)

Browse Source
This commit is contained in:
jschoubben
2026-10-23 15:22:32 +02:00
parent a48279b31a
commit 57258b6fa4
8 changed files with 82 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go
+5 -13
View File
@@ -1,11 +1,8 @@
package main
import (
"encoding/json"
"slices"
"strings"
"git.novox.be/novox/mesh-sdk/go/asks"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -15,11 +12,6 @@ import (
// which alone fills an ask's answer, takes it from there as it takes a warrant from the phone: under the ask's own key, once,
// by compare-and-set. The merge's ask needs proof (ADR 0421 §1, issue 953) and is never answered from a view key.
// An ask of the assigner as the teller reads it: the router's ask under "ask". Its other fields are the assigner's.
type assignerAsk struct {
Ask asks.Ask `json:"ask"`
}
// OnViewAnswer handles a key put in the view's bucket answers. A key that is no view answer key, or is the answer to an ask that
// needs proof of who answered, is left and said.
func (t *Teller) OnViewAnswer(key string) {
@@ -76,13 +68,13 @@ func (t *Teller) fillFromView(askKey string) {
t.logf("%s: the view's answer is not readable and fills nothing: %v", viewKey, err)
return
}
var ask assignerAsk
if err := json.Unmarshal(rawAsk, &ask); err != nil || ask.Ask.ID == "" {
t.logf("%s: the assigner's ask is not readable and takes no answer (%v)", askKey, err)
ask, err := delivery.DecodeAssignerAsk(askKey, rawAsk)
if err != nil {
t.logf("%s: the assigner's ask is not readable and takes no answer: %v", askKey, err)
return
}
if !slices.ContainsFunc(ask.Ask.Options, func(o asks.Option) bool { return o.ID == va.Option }) {
t.logf("%s: the view answered %q and the ask offers no option %q: nothing is written", askKey, va.Option, va.Option)
if err := delivery.ViewAnswerFits(va, ask); err != nil {
t.logf("%s: nothing is written: %v", askKey, err)
return
}
akey := delivery.AnswerKey(askKey)
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go
+25 -4
View File
@@ -1,7 +1,6 @@
package main
import (
"encoding/json"
"strings"
"testing"
@@ -22,16 +21,21 @@ func holdAsk(t *testing.T) (key string, value []byte, ask asks.Ask) {
ask = asks.Ask{ID: delivery.AskID(key), Headline: "plex is held after a failed health check", Explanation: "ace failed the check.", Who: asks.Operator,
Expires: now.Add(asks.ApproveLasts), OnExpiry: "the fingerprint stays held",
Options: []asks.Option{{ID: "release", Label: "Release", Does: "lets the move go on", Level: asks.Approve}, {ID: "keep", Label: "Keep held", Does: "keeps it held", Level: asks.Approve}}}
value, err = json.Marshal(map[string]any{"v": 1, "ask": ask})
value, err = delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: ask})
if err != nil {
t.Fatal(err)
}
return key, value, ask
}
func holdKey() string {
k, _ := delivery.HoldAskKey("plex", fpHeld)
return k
}
func viewAnswer(t *testing.T, option, words string) []byte {
t.Helper()
v, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Option: option, Words: words})
v, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Asking: delivery.AskID(holdKey()), Option: option, Words: words})
if err != nil {
t.Fatal(err)
}
@@ -108,7 +112,7 @@ func TestAViewAnswerOfAnOptionTheAskDoesNotOfferFillsNothing(t *testing.T) { //
assigner[key] = value
view[key+".view"] = viewAnswer(t, "delete", "")
b.t.OnViewAnswer(key + ".view")
if b.answers.writes != 0 || !strings.Contains(b.logs.String(), `offers no option "delete"`) {
if b.answers.writes != 0 || !strings.Contains(b.logs.String(), `chose "delete"`) {
t.Errorf("writes %d, log:\n%s", b.answers.writes, b.logs.String())
}
}
@@ -153,3 +157,20 @@ func TestAnUnreadableViewAnswerOrAskFillsNothing(t *testing.T) { // 1297/B10
t.Errorf("filled from something unreadable; log:\n%s", b.logs.String())
}
}
func TestAStaleViewAnswerOfAnotherAskingFillsNothing(t *testing.T) { // 1297/B10 (stale answer)
b := newBed(t)
assigner, view := withView(b)
key, value, _ := holdAsk(t)
assigner[key] = value
stale, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Asking: "an-earlier-asking", Option: "release"})
if err != nil {
t.Fatal(err)
}
view[key+".view"] = stale
b.t.OnViewAnswer(key + ".view")
b.t.OnAssignerAsk(key)
if b.answers.writes != 0 || !strings.Contains(b.logs.String(), "an-earlier-asking") {
t.Errorf("writes %d; log:\n%s", b.answers.writes, b.logs.String())
}
}
modules/mesh-delivery-teller/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-teller
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023151240-d9384bcb239a
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023132058-f625d98713e4
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-delivery-view/cmd/mesh-delivery-cli/main_test.go
+1 -1
View File
@@ -199,7 +199,7 @@ func TestTheAnswerCommandAnswersAnAskNeedingNoProof(t *testing.T) { // 1297/B9b
}
hold := asks.Ask{ID: delivery.AskID(key), Headline: "plex is held", Explanation: "ace failed the check.", Who: asks.Operator, Expires: time.Now().Add(time.Hour), OnExpiry: "stays held",
Options: []asks.Option{{ID: "release", Label: "Release", Does: "goes on", Level: asks.Approve}, {ID: "keep", Label: "Keep", Does: "stays", Level: asks.Approve}}}
value, _ := json.Marshal(map[string]any{"v": 1, "ask": hold})
value, _ := delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: hold})
heads := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "0123456789abcdef0123456789abcdef01234567"}}
mkey, err := delivery.MergeAskKey("feat/x", heads)
if err != nil {
modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go
+1 -1
View File
@@ -538,7 +538,7 @@ func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 129
ask := asks.Ask{ID: delivery.AskID(key), Headline: "plex is held after a failed health check", Explanation: "ace failed the check.", Who: asks.Operator,
Expires: time.Now().Add(time.Hour), OnExpiry: "the fingerprint stays held",
Options: []asks.Option{{ID: "release", Label: "Release", Does: "lets the move go on", Level: asks.Approve}, {ID: "keep", Label: "Keep held", Does: "keeps it held", Level: asks.Approve}}}
value, _ := json.Marshal(map[string]any{"v": 1, "ask": ask})
value, _ := delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: ask})
if _, err := askBucket.Put(ctx, key, value); err != nil {
t.Fatal(err)
}
modules/mesh-delivery-view/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-view
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023151240-d9384bcb239a
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023132058-f625d98713e4
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-delivery-view/internal/view/answers_test.go
+24 -2
View File
@@ -2,7 +2,6 @@ package view
import (
"bufio"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
@@ -43,7 +42,7 @@ func holdAsk(t *testing.T) (Entry, asks.Ask) {
a := asks.Ask{ID: delivery.AskID(key), Headline: "plex is held after a failed health check", Explanation: "ace failed the check on the new build.", Who: asks.Operator,
Expires: t0.Add(23 * time.Hour), OnExpiry: "the fingerprint stays held",
Options: []asks.Option{{ID: "release", Label: "Release", Does: "lets the move go on", Level: asks.Approve}, {ID: "keep", Label: "Keep held", Does: "keeps it held", Level: asks.Approve}}}
v, err := json.Marshal(map[string]any{"v": 1, "ask": a})
v, err := delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: a})
if err != nil {
t.Fatal(err)
}
@@ -215,3 +214,26 @@ func TestThePageAnswersAnAskAndRefusesAForeignPost(t *testing.T) { // 1297/B9a
}
}
}
func TestAStaleViewAnswerOfAnotherAskingIsNotTheChoice(t *testing.T) { // 1297/B9 (stale answer)
m := New()
e, a := holdAsk(t)
fill(m, e)
stale, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Asking: "an-earlier-asking", Option: "release"})
if err != nil {
t.Fatal(err)
}
fill(m, Entry{Bucket: delivery.ViewAnswersBucket, Key: e.Key + ".view", Value: stale, At: t0})
if got := m.Snapshot().Asks; len(got) != 1 || got[0].Chosen != "" {
t.Errorf("a stale answer shows as the choice: %+v", got)
}
// and the view may answer this asking, writing its id
w := &memWriter{}
m.SetWriter(w)
if err := m.Answer(e.Key, "keep", ""); err != nil {
t.Fatal(err)
}
if v, err := delivery.DecodeViewAnswer(w.puts[e.Key+".view"]); err != nil || v.Asking != a.ID {
t.Errorf("answer = %+v, %v; want asking %s", v, err, a.ID)
}
}
modules/mesh-delivery-view/internal/view/view.go
+24 -29
View File
@@ -177,14 +177,14 @@ type UnreadableRow struct {
type Model struct {
mu sync.Mutex
assigns map[string]assignment
builds map[string]buildEntry // by <module>.build.<id>
groups map[string]groupEntry // the indexer's candidates, by group
modules map[string]moduleEntry // the indexer's modules, by module
keys map[string]keyEntry // the builder's build keys, by <module>.<fingerprint>
asks map[string]askRecord // the gate's asks, by key
diffs map[string]diffEntry // the assigner's differences, by <node>.<module>
aasks map[string]assignerAskRecord // the assigner's asks (hold, restore), by key
chosen map[string]string // the options the view wrote, by ask key
builds map[string]buildEntry // by <module>.build.<id>
groups map[string]groupEntry // the indexer's candidates, by group
modules map[string]moduleEntry // the indexer's modules, by module
keys map[string]keyEntry // the builder's build keys, by <module>.<fingerprint>
asks map[string]askRecord // the gate's asks, by key
diffs map[string]diffEntry // the assigner's differences, by <node>.<module>
aasks map[string]assignerAskRecord // the assigner's asks (hold, restore), by key
chosen map[string]delivery.ViewAnswer // the answers the view wrote, by ask key
writer Writer
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
@@ -286,7 +286,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{}, diffs: map[string]diffEntry{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]string{}, 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{}, diffs: map[string]diffEntry{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]delivery.ViewAnswer{}, answers: map[string]string{}, newest: map[string]string{},
haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{},
presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}}
}
@@ -379,7 +379,7 @@ func (m *Model) Apply(e Entry) {
return
}
if a, err := delivery.DecodeViewAnswer(e.Value); err == nil {
m.chosen[askKey] = a.Option
m.chosen[askKey] = a
}
case delivery.AssignerDifferencesBucket:
if e.Deleted {
@@ -737,7 +737,7 @@ func (m *Model) askRows() []AskRow {
continue
}
row := AskRow{Key: key, Headline: a.ask.Headline, Words: a.ask.Explanation, Expires: a.ask.Expires, How: AnswerInTheView, Pulls: []string{},
Chosen: m.chosen[key]}
Chosen: m.chosenOf(key, a.ask)}
for _, o := range a.ask.Options {
row.Options = append(row.Options, o.ID)
}
@@ -747,27 +747,22 @@ func (m *Model) askRows() []AskRow {
return out
}
// decodeAssignerAsk reads a key of the assigner's asks: the router's ask under "ask" (`{"v":1,"ask":{…}}`, the gate's ask
// without its group and heads; the layout of the assigner's asks is the assigner's, issue 1297 B5). The key must be a kind of
// ask that needs no proof, and the ask must offer an option.
// decodeAssignerAsk reads a key of the assigner's asks (mesh-sdk go/delivery DecodeAssignerAsk): the router's ask under its
// asking's id, of a kind that needs no proof, offering an option.
func decodeAssignerAsk(key string, value []byte) (asks.Ask, error) {
k, err := delivery.ParseAnyAskKey(key)
a, err := delivery.DecodeAssignerAsk(key, value)
if err != nil {
return asks.Ask{}, err
}
if proof, _ := delivery.AskNeedsProof(k.Kind); proof {
return asks.Ask{}, fmt.Errorf("%s needs proof of who answered and is no ask of the assigner", key)
}
var v struct {
Ask asks.Ask `json:"ask"`
}
if err := json.Unmarshal(value, &v); err != nil {
return asks.Ask{}, fmt.Errorf("the assigner's ask %s is not readable: %w", key, err)
}
if v.Ask.ID == "" || len(v.Ask.Options) == 0 {
return asks.Ask{}, fmt.Errorf("the assigner's ask %s names no id or no option", key)
return a.Ask, nil
}
// chosenOf is the option the view wrote for this asking of the ask: an answer to another asking (a stale one) is none.
func (m *Model) chosenOf(key string, ask asks.Ask) string {
if a, ok := m.chosen[key]; ok && a.Asking == ask.ID {
return a.Option
}
return v.Ask, nil
return ""
}
// Writer puts a key of the view's bucket answers, the only state the view writes.
@@ -797,7 +792,7 @@ func (m *Model) Answer(askKey, option, words string) error {
m.mu.Lock()
a, held := m.aasks[askKey]
open := held && a.err == "" && m.answers[askKey] != a.ask.ID
chosen, w := m.chosen[askKey], m.writer
chosen, w := m.chosenOf(askKey, a.ask), m.writer
m.mu.Unlock()
if !open {
return fmt.Errorf("the view holds no open ask %s", askKey)
@@ -815,7 +810,7 @@ func (m *Model) Answer(askKey, option, words string) error {
if w == nil {
return fmt.Errorf("the view cannot write its answers here")
}
body, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Option: option, Words: strings.TrimSpace(words)})
body, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Asking: a.ask.ID, Option: option, Words: strings.TrimSpace(words)})
if err != nil {
return err
}