Merge branch 'feat/r4g-view' into feat/r4g
This commit is contained in:
@@ -165,6 +165,8 @@ const (
|
||||
tellerAnswers = "mesh-delivery-teller_answers"
|
||||
tellerAsked = "mesh-delivery-teller_asked"
|
||||
gateAsksKV = "mesh-delivery-gate_asks"
|
||||
assignerAsks = "mesh-delivery-assigner_asks"
|
||||
viewAnswers = "mesh-delivery-view_answers"
|
||||
seatAsk = "mesh.seat.operator-channel.accept.ask.mesh-delivery-teller"
|
||||
seatCancel = "mesh.seat.operator-channel.accept.cancel.mesh-delivery-teller"
|
||||
seatCondition = "mesh.seat.operator-channel.accept.condition.mesh-delivery-teller"
|
||||
@@ -174,7 +176,9 @@ const (
|
||||
func fullConf(dir string) string {
|
||||
teller := append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), readOf(gateAsksKV)...),
|
||||
seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_asks.$KV.asks.mesh-delivery-teller.>")
|
||||
view := append(readOf(tellerAnswers), readOf(gateAsksKV)...)
|
||||
teller = append(append(teller, readOf(assignerAsks)...), readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads
|
||||
view := append(append(append(readOf(tellerAnswers), readOf(gateAsksKV)...), readOf(assignerAsks)...), keepsOf(viewAnswers)...)
|
||||
assigner := keepsOf(assignerAsks)
|
||||
tools := append(keepsOf(gateAsksKV), readOf(tellerAnswers)...)
|
||||
return fmt.Sprintf(`
|
||||
listen: 127.0.0.1:-1
|
||||
@@ -184,12 +188,13 @@ accounts {
|
||||
jetstream: enabled
|
||||
users = [
|
||||
{ user: "ctl", password: "pw" }
|
||||
%s%s%s ]
|
||||
%s%s%s%s ]
|
||||
}
|
||||
}
|
||||
`, filepath.Join(dir, "js"),
|
||||
user("n1.mesh-delivery-teller", teller, []string{seatDecided}),
|
||||
user("n1.mesh-delivery-view", view, nil),
|
||||
user("n1.mesh-delivery-assigner", assigner, nil),
|
||||
user("n1.node-tools", tools, nil))
|
||||
}
|
||||
|
||||
@@ -238,7 +243,7 @@ func startFull(t *testing.T) *fullBed {
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
t.Cleanup(cancel)
|
||||
for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1} {
|
||||
for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1, assignerAsks: delivery.AssignerAsksHistory, viewAnswers: delivery.ViewAnswersHistory} {
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -468,3 +473,75 @@ func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
|
||||
say(asks.Warrant{Ask: ga2.Ask.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeExpired, At: now, AskDigest: ga2.Ask.Digest()})
|
||||
lastAsk(held.Ask.ID, second)
|
||||
}
|
||||
|
||||
// TestTellerFillsAnAskFromTheViewsAnswerOnce starts the program against the bus with the permissions above: the assigner's
|
||||
// fixture hold ask is written, then the view's answer under `<ask key>.view`; the answer is under the ask's own key, once
|
||||
// (rewriting the view's answer does not write a second), and a view answer under the merge's key fills nothing
|
||||
// (1297/B10 AC: a view answer fills the ask once; an ask needing proof is not answered from a view key).
|
||||
func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
|
||||
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)
|
||||
}
|
||||
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) })
|
||||
|
||||
key, value, ask := holdAsk(t)
|
||||
askKV, err := assigner.KeyValue(b.ctx, assignerAsks)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
viewKV, err := view.KeyValue(b.ctx, viewAnswers)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
answers, err := b.js.KeyValue(b.ctx, tellerAnswers)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(500 * time.Millisecond) // the program is watching
|
||||
if _, err := askKV.Put(b.ctx, key, value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := viewKV.Put(b.ctx, key+".view", viewAnswer(t, "release", "the check was a flake")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var e jetstream.KeyValueEntry
|
||||
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 {
|
||||
break
|
||||
}
|
||||
}
|
||||
if e == nil {
|
||||
t.Fatalf("no answer under %s: %v\n%s", key, err, logs.String())
|
||||
}
|
||||
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 {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
if again, err := answers.Get(b.ctx, delivery.AnswerKey(key)); err != nil || again.Revision() != e.Revision() {
|
||||
t.Errorf("the answer was written a second time: %v", err)
|
||||
}
|
||||
// An answer under the merge's key fills nothing.
|
||||
mkey, mvalue, _ := gateAsk(t, twoHeads(), twoHeads())
|
||||
if _, err := askKV.Put(b.ctx, mkey, mvalue); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := viewKV.Put(b.ctx, mkey+".view", viewAnswer(t, "approve", "")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
if _, err := answers.Get(b.ctx, delivery.AnswerKey(mkey)); err == nil {
|
||||
t.Error("the merge's ask was answered from a view key")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -137,10 +137,43 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err := client.Listen(t.Answer, t.Said); err != nil {
|
||||
return fmt.Errorf("hearing the router's warrants: %w", err)
|
||||
}
|
||||
if err := watchAsks(ctx, gate, t.OnAsk, t.OnAskGone, t.OnCurrent); err != nil {
|
||||
return fmt.Errorf("watching the gate's asks: %w", err)
|
||||
// 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)
|
||||
assignerKV, viewKV := optionalBucket(ctx, js, delivery.AssignerAsksBucket, logger), optionalBucket(ctx, js, delivery.ViewAnswersBucket, logger)
|
||||
if assignerKV != nil && viewKV != nil {
|
||||
t.Assigner, t.View = kvGate{ctx: ctx, kv: assignerKV}, kvGate{ctx: ctx, kv: viewKV}
|
||||
go func() {
|
||||
failed <- watchAsks(ctx, viewKV, func(key string, _ uint64, _ []byte) { t.OnViewAnswer(key) }, func(string) {}, nil)
|
||||
}()
|
||||
go func() {
|
||||
failed <- watchAsks(ctx, assignerKV, func(key string, _ uint64, _ []byte) { t.OnAssignerAsk(key) }, func(string) {}, nil)
|
||||
}()
|
||||
}
|
||||
return nil
|
||||
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)
|
||||
}
|
||||
}()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case err := <-failed:
|
||||
if err != nil {
|
||||
return fmt.Errorf("a watch ended: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// optionalBucket is a bucket the teller reads when it exists, and nil, said once, when it does not.
|
||||
func optionalBucket(ctx context.Context, js jetstream.JetStream, name string, logger *log.Logger) jetstream.KeyValue {
|
||||
kv, err := js.KeyValue(ctx, name)
|
||||
if err != nil {
|
||||
logger.Printf("the bucket %s cannot be read (%v): the view's answers fill no ask until the teller is started again with it", name, err)
|
||||
return nil
|
||||
}
|
||||
return kv
|
||||
}
|
||||
|
||||
// askedBucket is the teller's state bucket asked, `<module>_<name>`: what it asked of the router and which condition it
|
||||
|
||||
@@ -58,6 +58,10 @@ type Teller struct {
|
||||
Store store
|
||||
Answers answers
|
||||
Gate gateAsks
|
||||
// Assigner is the assigner's bucket asks and View the view's bucket answers (issue 1297 B10); with either nil the teller
|
||||
// answers nothing from the view.
|
||||
Assigner gateAsks
|
||||
View gateAsks
|
||||
// Say publishes a condition to the operator (mesh.seat.operator-channel.accept.condition.<module>).
|
||||
Say func(c asks.Condition) error
|
||||
Logf func(format string, args ...any)
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The view's answers (novox/hq ADR 0528 clause 6, ADR 0437 §10, issue 1297 B10). The operator may answer, in the view, the
|
||||
// asks of the assigner that need no proof of who answered (the hold of a fingerprint and the restore of a data module set back,
|
||||
// mesh-sdk go/delivery AskNeedsProof); the view writes its answer under `<ask key>.view` in its bucket answers, and the teller,
|
||||
// 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.
|
||||
|
||||
// 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) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
askKey, err := delivery.ParseViewAnswerKey(key)
|
||||
if err != nil {
|
||||
t.logf("the view's key %s is left: %v", key, err)
|
||||
return
|
||||
}
|
||||
t.fillFromView(askKey)
|
||||
}
|
||||
|
||||
// OnAssignerAsk handles a key put in the assigner's bucket asks: a view answer that came before its ask fills it now.
|
||||
func (t *Teller) OnAssignerAsk(key string) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
if _, err := delivery.ViewAnswerKey(key); err != nil {
|
||||
t.logf("the assigner's key %s is left and never answered from the view: %v", key, err)
|
||||
return
|
||||
}
|
||||
t.fillFromView(key)
|
||||
}
|
||||
|
||||
// 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.
|
||||
func (t *Teller) fillFromView(askKey string) {
|
||||
if t.Assigner == nil || t.View == nil {
|
||||
return
|
||||
}
|
||||
viewKey, err := delivery.ViewAnswerKey(askKey) // refuses an ask that needs proof, naming the phone
|
||||
if err != nil {
|
||||
t.logf("%s is not answered from the view: %v", askKey, err)
|
||||
return
|
||||
}
|
||||
rawAnswer, found, err := t.View.Get(viewKey)
|
||||
if err != nil {
|
||||
t.logf("%s: the view's answer could not be read, so the ask is left: %v", askKey, err)
|
||||
return
|
||||
}
|
||||
if !found {
|
||||
return
|
||||
}
|
||||
rawAsk, found, err := t.Assigner.Get(askKey)
|
||||
if err != nil {
|
||||
t.logf("%s: the assigner's ask could not be read, so the view's answer is left: %v", askKey, err)
|
||||
return
|
||||
}
|
||||
if !found {
|
||||
return // the answer waits for its ask: OnAssignerAsk fills it when it comes
|
||||
}
|
||||
va, err := delivery.DecodeViewAnswer(rawAnswer)
|
||||
if err != nil {
|
||||
t.logf("%s: the view's answer is not readable and fills nothing: %v", viewKey, err)
|
||||
return
|
||||
}
|
||||
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 err := delivery.ViewAnswerFits(va, ask); err != nil {
|
||||
t.logf("%s: nothing is written: %v", askKey, err)
|
||||
return
|
||||
}
|
||||
akey := delivery.AnswerKey(askKey)
|
||||
cur, crev, has, err := t.Answers.Get(akey)
|
||||
if err != nil {
|
||||
t.logf("%s: the answer could not be read: %v", askKey, err)
|
||||
return
|
||||
}
|
||||
if has && answeredAsking(cur) == ask.Ask.ID {
|
||||
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})
|
||||
if err != nil {
|
||||
t.logf("%s: the answer could not be made: %v", askKey, err)
|
||||
return
|
||||
}
|
||||
written, err := t.Answers.Write(akey, body, crev, has)
|
||||
switch {
|
||||
case err != nil:
|
||||
t.logf("%s: the answer could not be written: %v", askKey, err)
|
||||
case !written:
|
||||
t.logf("%s: another writer answered first, and the view's answer is not written", askKey)
|
||||
default:
|
||||
t.logf("%s: %s", askKey, words)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,176 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
const fpHeld = "src2:2222222222222222222222222222222222222222222222222222222222222222"
|
||||
|
||||
// holdAsk is a key of the assigner's asks: the ask about a held fingerprint, as a fixture (in release 5 the assigner writes
|
||||
// none yet; the value is `{"v":1,"ask":<the router's ask>}`).
|
||||
func holdAsk(t *testing.T) (key string, value []byte, ask asks.Ask) {
|
||||
t.Helper()
|
||||
key, err := delivery.HoldAskKey("plex", fpHeld)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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 = 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{Asking: delivery.AskID(holdKey()), Option: option, Words: words})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
// withView gives the bed's teller the assigner's asks and the view's answers, both empty.
|
||||
func withView(b *bed) (assigner, view memGate) {
|
||||
assigner, view = memGate{}, memGate{}
|
||||
b.t.Assigner, b.t.View = assigner, view
|
||||
return assigner, view
|
||||
}
|
||||
|
||||
func TestAViewAnswerFillsTheAskOnce(t *testing.T) { // 1297/B10 AC
|
||||
b := newBed(t)
|
||||
assigner, view := withView(b)
|
||||
key, value, ask := holdAsk(t)
|
||||
assigner[key] = value
|
||||
view[key+".view"] = viewAnswer(t, "release", "the check was a flake")
|
||||
b.t.OnViewAnswer(key + ".view")
|
||||
raw, ok := b.answers.written[key]
|
||||
if !ok {
|
||||
t.Fatalf("no answer under %s; log:\n%s", key, b.logs.String())
|
||||
}
|
||||
a, err := delivery.DecodeAnswer(raw)
|
||||
if err != nil || a.Option != "release" || answeredAsking(raw) != ask.ID || !strings.Contains(a.Words, "view") || !strings.Contains(a.Words, "the check was a flake") {
|
||||
t.Errorf("answer = %+v, %v", a, err)
|
||||
}
|
||||
// The same answer heard again (the watch replays, the ask is rewritten) writes nothing more.
|
||||
b.t.OnViewAnswer(key + ".view")
|
||||
b.t.OnAssignerAsk(key)
|
||||
if b.answers.writes != 1 {
|
||||
t.Errorf("%d writes, want one", b.answers.writes)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAViewAnswerHeardBeforeItsAskFillsItWhenTheAskComes(t *testing.T) { // 1297/B10
|
||||
b := newBed(t)
|
||||
assigner, view := withView(b)
|
||||
key, value, _ := holdAsk(t)
|
||||
view[key+".view"] = viewAnswer(t, "keep", "")
|
||||
b.t.OnViewAnswer(key + ".view")
|
||||
if b.answers.writes != 0 {
|
||||
t.Fatalf("answered an ask the teller does not hold")
|
||||
}
|
||||
assigner[key] = value
|
||||
b.t.OnAssignerAsk(key)
|
||||
if a, err := delivery.DecodeAnswer(b.answers.written[key]); err != nil || a.Option != "keep" {
|
||||
t.Errorf("answer = %+v, %v", a, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnAskNeedingProofIsNotAnsweredFromAViewKey(t *testing.T) { // 1297/B10 AC
|
||||
b := newBed(t)
|
||||
assigner, view := withView(b)
|
||||
key, value, _ := gateAsk(t, twoHeads(), twoHeads())
|
||||
// Even with the merge's ask under the assigner's bucket and a view answer under its key, nothing is written.
|
||||
assigner[key] = value
|
||||
view[key+".view"] = viewAnswer(t, "approve", "")
|
||||
b.t.OnViewAnswer(key + ".view")
|
||||
b.t.OnAssignerAsk(key)
|
||||
if b.answers.writes != 0 || len(b.answers.written) != 0 {
|
||||
t.Fatalf("the merge's ask was answered from the view: %v", b.answers.written)
|
||||
}
|
||||
if !strings.Contains(b.logs.String(), "issue 953") {
|
||||
t.Errorf("the refusal does not say why:\n%s", b.logs.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAViewAnswerOfAnOptionTheAskDoesNotOfferFillsNothing(t *testing.T) { // 1297/B10
|
||||
b := newBed(t)
|
||||
assigner, view := withView(b)
|
||||
key, value, _ := holdAsk(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(), `chose "delete"`) {
|
||||
t.Errorf("writes %d, log:\n%s", b.answers.writes, b.logs.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAViewAnswerNeverReplacesTheAnswerOfTheSameAsking(t *testing.T) { // 1297/B10
|
||||
b := newBed(t)
|
||||
assigner, view := withView(b)
|
||||
key, value, ask := holdAsk(t)
|
||||
assigner[key] = value
|
||||
view[key+".view"] = viewAnswer(t, "keep", "")
|
||||
early, err := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: ask.ID + "@2026-10-20T09:00:00Z", Words: "answered elsewhere"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.answers.written[key], b.answers.revs[key] = early, 7
|
||||
b.t.OnViewAnswer(key + ".view")
|
||||
if b.answers.writes != 0 || string(b.answers.written[key]) != string(early) {
|
||||
t.Errorf("the answer of the same asking was replaced")
|
||||
}
|
||||
// An earlier asking's answer (another router id) does not stop this one.
|
||||
older, _ := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: "an-older-id@2026-10-19T09:00:00Z", Words: "old"})
|
||||
b.answers.written[key], b.answers.revs[key] = older, 8
|
||||
b.t.OnViewAnswer(key + ".view")
|
||||
if b.answers.writes != 1 || answeredAsking(b.answers.written[key]) != ask.ID {
|
||||
t.Errorf("writes %d, answer %s", b.answers.writes, b.answers.written[key])
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnUnreadableViewAnswerOrAskFillsNothing(t *testing.T) { // 1297/B10
|
||||
b := newBed(t)
|
||||
assigner, view := withView(b)
|
||||
key, value, _ := holdAsk(t)
|
||||
assigner[key] = []byte(`{"v":1}`)
|
||||
view[key+".view"] = viewAnswer(t, "release", "")
|
||||
b.t.OnViewAnswer(key + ".view")
|
||||
assigner[key] = value
|
||||
view[key+".view"] = []byte(`{"v":99}`)
|
||||
b.t.OnViewAnswer(key + ".view")
|
||||
b.t.OnViewAnswer("not-a-view-key")
|
||||
b.t.OnViewAnswer("hold.x.view")
|
||||
if b.answers.writes != 0 {
|
||||
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())
|
||||
}
|
||||
}
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-teller
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261020060201-88581d0acaef
|
||||
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
|
||||
)
|
||||
|
||||
@@ -81,7 +81,8 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1
|
||||
if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}}; !reflect.DeepEqual(m.State, want) {
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
if want := []string{"mesh-delivery-gate.asks"}; !reflect.DeepEqual(sorted(m.Reads), 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).
|
||||
if want := []string{"mesh-delivery-assigner.asks", "mesh-delivery-gate.asks", "mesh-delivery-view.answers"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
if want := []string{"operator-channel"}; !reflect.DeepEqual(sorted(m.Uses), want) {
|
||||
|
||||
@@ -7,7 +7,9 @@
|
||||
{ "name": "asked", "history": 1 }
|
||||
],
|
||||
"reads": [
|
||||
"mesh-delivery-gate.asks"
|
||||
"mesh-delivery-assigner.asks",
|
||||
"mesh-delivery-gate.asks",
|
||||
"mesh-delivery-view.answers"
|
||||
],
|
||||
"uses": [
|
||||
"operator-channel"
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
// 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 and changes nothing.
|
||||
// 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).
|
||||
package main
|
||||
|
||||
import (
|
||||
@@ -8,6 +9,7 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"mesh-delivery-view/internal/view"
|
||||
)
|
||||
@@ -30,8 +32,17 @@ func run(args []string, socket string, out, errOut io.Writer) int {
|
||||
return 2
|
||||
}
|
||||
if fs.NArg() > 0 {
|
||||
fmt.Fprintln(errOut, "mesh-delivery-cli takes no argument: it only shows")
|
||||
return 2
|
||||
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]")
|
||||
return 2
|
||||
}
|
||||
key, option := fs.Arg(1), fs.Arg(2)
|
||||
if err := view.AnswerAsk(socket, key, option, strings.Join(fs.Args()[3:], " ")); err != nil {
|
||||
fmt.Fprintln(errOut, "mesh-delivery-cli:", err)
|
||||
return 1
|
||||
}
|
||||
fmt.Fprintf(out, "answered %s with %s; the teller fills the ask from it\n", key, option)
|
||||
return 0
|
||||
}
|
||||
s, err := view.Ask(socket)
|
||||
if err != nil {
|
||||
|
||||
@@ -167,3 +167,76 @@ func TestPrintsTheOpenAsksInTheTableAndTheJSON(t *testing.T) { // 1345/AC2
|
||||
t.Errorf("json = %s (%v)", out.String(), err)
|
||||
}
|
||||
}
|
||||
|
||||
type putter struct{ puts map[string][]byte }
|
||||
|
||||
func (p *putter) Put(key string, value []byte) error { p.puts[key] = value; return nil }
|
||||
|
||||
func serveWith(t *testing.T, m *view.Model) string {
|
||||
t.Helper()
|
||||
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "c")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { os.RemoveAll(dir) })
|
||||
l, err := view.ListenSocket(filepath.Join(dir, "cli.sock"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
go view.ServeSocket(ctx, l, m)
|
||||
t.Cleanup(cancel)
|
||||
return filepath.Join(dir, "cli.sock")
|
||||
}
|
||||
|
||||
// TestTheAnswerCommandAnswersAnAskNeedingNoProof: `answer <ask key> <option> [words]` writes the view's answer; the merge's
|
||||
// ask is refused with the phone as the reason (1297/B9b).
|
||||
func TestTheAnswerCommandAnswersAnAskNeedingNoProof(t *testing.T) { // 1297/B9b AC
|
||||
const fp = "src2:2222222222222222222222222222222222222222222222222222222222222222"
|
||||
key, err := delivery.HoldAskKey("plex", fp)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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, _ := 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 {
|
||||
t.Fatal(err)
|
||||
}
|
||||
merge := asks.Ask{ID: delivery.AskID(mkey), Headline: "Merge", Explanation: "Merges.", Who: asks.Operator, Expires: time.Now().Add(time.Hour), OnExpiry: "held",
|
||||
Options: []asks.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: asks.Approve, Binds: "sha256:00"}}}
|
||||
mv, err := delivery.EncodeAsk(delivery.Ask{Group: "feat/x", Heads: heads, Ask: merge})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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.AsksBucket, Key: mkey, Value: mv})
|
||||
socket := serveWith(t, m)
|
||||
|
||||
var out, errOut bytes.Buffer
|
||||
if code := run([]string{"answer", key, "release", "the", "check", "was", "a", "flake"}, socket, &out, &errOut); code != 0 {
|
||||
t.Fatalf("exit %d: %s", code, errOut.String())
|
||||
}
|
||||
a, err := delivery.DecodeViewAnswer(w.puts[key+".view"])
|
||||
if err != nil || a.Option != "release" || a.Words != "the check was a flake" || !strings.Contains(out.String(), key) {
|
||||
t.Errorf("answer = %+v, %v; out %q", a, err, out.String())
|
||||
}
|
||||
out.Reset()
|
||||
errOut.Reset()
|
||||
if code := run([]string{"answer", mkey, "approve"}, socket, &out, &errOut); code != 1 || !strings.Contains(errOut.String(), "answer it on the phone (issue 953)") {
|
||||
t.Errorf("merge: exit %d, stderr %q", code, errOut.String())
|
||||
}
|
||||
if len(w.puts) != 1 {
|
||||
t.Errorf("puts = %v, want only the hold's", w.puts)
|
||||
}
|
||||
for _, args := range [][]string{{"answer"}, {"answer", key}, {"settle", key, "release"}} {
|
||||
errOut.Reset()
|
||||
if code := run(args, socket, &out, &errOut); code != 2 || errOut.Len() == 0 {
|
||||
t.Errorf("%v: exit %d, stderr %q", args, code, errOut.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -390,3 +391,214 @@ func TestProcessWholeShowsTheOpenAsksAndDropsTheAnsweredOnes(t *testing.T) { //
|
||||
}
|
||||
wait(func(s view.Snapshot) bool { return len(s.Asks) == 0 }, "the answered ask never left the list")
|
||||
}
|
||||
|
||||
// TestProcessWholeListsEveryDifference starts the program whole before the assigner's differences bucket exists, then
|
||||
// creates it and writes a difference explained by a rule and one unexplained: the socket's snapshot lists both with their
|
||||
// rule, and a difference taken away leaves the list (1297/B8: the view lists every difference with its rule).
|
||||
func TestProcessWholeListsEveryDifference(t *testing.T) { // 1297/B8 AC
|
||||
srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go srv.Start()
|
||||
if !srv.ReadyForConnections(10 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
t.Cleanup(srv.Shutdown)
|
||||
nc, err := nats.Connect(srv.ClientURL())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
js, err := jetstream.New(nc)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { os.RemoveAll(dir) })
|
||||
broker := filepath.Join(dir, "broker")
|
||||
cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
|
||||
if err := os.WriteFile(broker, cred, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
socket := filepath.Join(dir, "run", "cli.sock")
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
|
||||
runCtx, stop := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
|
||||
t.Cleanup(func() { stop(); <-done })
|
||||
|
||||
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerDifferencesBucket, History: delivery.AssignerDifferencesHistory})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
const fpOld = "src2:1111111111111111111111111111111111111111111111111111111111111111"
|
||||
const fpNew = "src2:2222222222222222222222222222222222222222222222222222222222222222"
|
||||
for key, d := range map[string]delivery.Difference{
|
||||
"ace.plex": {Want: fpOld, Sent: fpNew, Rule: delivery.RuleCanary, Words: "ace is not the canary"},
|
||||
"g14.plex": {Want: fpOld, Sent: fpNew, Unexplained: true},
|
||||
} {
|
||||
v, err := delivery.EncodeDifference(d)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := kv.Put(ctx, key, v); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
wait := func(check func(view.Snapshot) bool, what string) {
|
||||
t.Helper()
|
||||
var last view.Snapshot
|
||||
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
|
||||
if s, err := view.Ask(socket); err == nil {
|
||||
last = s
|
||||
if check(s) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
t.Fatalf("%s; last: %+v", what, last)
|
||||
}
|
||||
wait(func(s view.Snapshot) bool {
|
||||
d := s.Differences
|
||||
return len(d) == 2 && d[0].Node == "ace" && d[0].Rule == delivery.RuleCanary && d[1].Node == "g14" && d[1].Unexplained
|
||||
}, "the two differences never showed with their rules")
|
||||
if err := kv.Delete(ctx, "g14.plex"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wait(func(s view.Snapshot) bool { return len(s.Differences) == 1 }, "the difference taken away never left the list")
|
||||
}
|
||||
|
||||
// TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof starts the program whole, with the assigner's asks, the view's
|
||||
// answers and the teller's answers buckets, and a fixture hold ask (in release 5 the assigner writes none yet). The
|
||||
// operator's answer over the socket is written under `<ask key>.view` in the view's bucket; the ask shows the choice, and
|
||||
// leaves the list once the teller's answer is under its key. The merge's Approve is refused with its reason and writes
|
||||
// nothing (1297/B9a AC).
|
||||
func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 1297/B9a AC
|
||||
srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go srv.Start()
|
||||
if !srv.ReadyForConnections(10 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
t.Cleanup(srv.Shutdown)
|
||||
nc, err := nats.Connect(srv.ClientURL())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
js, err := jetstream.New(nc)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
bucket := func(name string, history uint8) jetstream.KeyValue {
|
||||
t.Helper()
|
||||
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return kv
|
||||
}
|
||||
askBucket := bucket(delivery.AssignerAsksBucket, delivery.AssignerAsksHistory)
|
||||
viewBucket := bucket(delivery.ViewAnswersBucket, delivery.ViewAnswersHistory)
|
||||
teller := bucket(delivery.AnswersBucket, delivery.AnswersHistory)
|
||||
gate := bucket(delivery.AsksBucket, delivery.AsksHistory)
|
||||
|
||||
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { os.RemoveAll(dir) })
|
||||
broker := filepath.Join(dir, "broker")
|
||||
cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
|
||||
if err := os.WriteFile(broker, cred, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
socket := filepath.Join(dir, "run", "cli.sock")
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
|
||||
runCtx, stop := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
|
||||
t.Cleanup(func() { stop(); <-done })
|
||||
|
||||
const fp = "src2:2222222222222222222222222222222222222222222222222222222222222222"
|
||||
key, err := delivery.HoldAskKey("plex", fp)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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, _ := delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: ask})
|
||||
if _, err := askBucket.Put(ctx, key, value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wait := func(check func(view.Snapshot) bool, what string) {
|
||||
t.Helper()
|
||||
var last view.Snapshot
|
||||
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
|
||||
if s, err := view.Ask(socket); err == nil {
|
||||
last = s
|
||||
if check(s) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
t.Fatalf("%s; last: %+v", what, last)
|
||||
}
|
||||
wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].How == view.AnswerInTheView }, "the open ask never showed")
|
||||
|
||||
if err := view.AnswerAsk(socket, key, "release", "the check was a flake"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
e, err := viewBucket.Get(ctx, key+".view")
|
||||
if err != nil {
|
||||
t.Fatalf("the view wrote no answer under %s.view: %v", key, err)
|
||||
}
|
||||
if a, err := delivery.DecodeViewAnswer(e.Value()); err != nil || a.Option != "release" {
|
||||
t.Fatalf("answer = %+v, %v", a, err)
|
||||
}
|
||||
wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].Chosen == "release" }, "the ask never showed the view's choice")
|
||||
|
||||
// The merge's Approve is answered on the phone alone.
|
||||
heads := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "0123456789abcdef0123456789abcdef01234567"}}
|
||||
mkey, err := delivery.MergeAskKey("feat/x", heads)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
merge := asks.Ask{ID: delivery.AskID(mkey), Headline: "Merge", Explanation: "Merges.", Who: asks.Operator, Expires: time.Now().Add(time.Hour), OnExpiry: "held",
|
||||
Options: []asks.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: asks.Approve, Binds: "sha256:00"}}}
|
||||
mv, err := delivery.EncodeAsk(delivery.Ask{Group: "feat/x", Heads: heads, Ask: merge})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := gate.Put(ctx, mkey, mv); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wait(func(s view.Snapshot) bool { return len(s.Asks) == 2 }, "the merge ask never showed")
|
||||
if err := view.AnswerAsk(socket, mkey, "approve", ""); err == nil || !strings.Contains(err.Error(), "answer it on the phone (issue 953)") {
|
||||
t.Errorf("merge Approve = %v, want the refusal naming the phone", err)
|
||||
}
|
||||
if _, err := viewBucket.Get(ctx, mkey+".view"); err == nil {
|
||||
t.Error("the view wrote an answer to the merge's ask")
|
||||
}
|
||||
|
||||
// The teller fills the ask: it leaves the list.
|
||||
answer, err := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: ask.ID + "@2026-10-23T12:00:00Z", Words: "answered in the view"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := teller.Put(ctx, key, answer); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].Key == mkey }, "the answered ask never left the list")
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
// Command mesh-delivery-view is the operator's view of delivery (novox/hq ADR 0429 §3): it watches the have, assignments
|
||||
// and builds buckets on its own bus credential, derives one row per module and node in memory, and serves them on a
|
||||
// local socket (for mesh-delivery-cli) and on a local web page. It writes nothing and starts, stops or retries nothing.
|
||||
// local socket (for mesh-delivery-cli) and on a local web page. It starts, stops or retries nothing; the one thing it writes is the operator's answer to an ask that needs no proof of who
|
||||
// answered, under `<ask key>.view` in its bucket answers (ADR 0528 clause 6).
|
||||
package main
|
||||
|
||||
import (
|
||||
@@ -136,7 +137,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket} {
|
||||
model.SetWriter(kvWriter{ctx: ctx, js: js})
|
||||
for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket} {
|
||||
go follow(ctx, js, bucket, model, logger)
|
||||
}
|
||||
select {
|
||||
@@ -152,6 +154,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return nil
|
||||
}
|
||||
|
||||
// kvWriter is the view's one write: a key of its own bucket answers (manifest state entry answers), which the teller reads.
|
||||
// The bucket is the controller's to create; a bucket not there yet is an error the operator is told.
|
||||
type kvWriter struct {
|
||||
ctx context.Context
|
||||
js jetstream.JetStream
|
||||
}
|
||||
|
||||
func (w kvWriter) Put(key string, value []byte) error {
|
||||
kv, err := w.js.KeyValue(w.ctx, delivery.ViewAnswersBucket)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the bucket %s: %w", delivery.ViewAnswersBucket, err)
|
||||
}
|
||||
_, err = kv.Put(w.ctx, key, value)
|
||||
return err
|
||||
}
|
||||
|
||||
// follow watches a bucket with its last values first and then each change, for as long as ctx lives: a bucket that is
|
||||
// not there yet, or a watch that ends, is tried again.
|
||||
func follow(ctx context.Context, js jetstream.JetStream, bucket string, model *view.Model, logger *log.Logger) {
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-view
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
|
||||
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
|
||||
)
|
||||
|
||||
@@ -0,0 +1,239 @@
|
||||
package view
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// memWriter is the view's bucket answers in memory.
|
||||
type memWriter struct {
|
||||
puts map[string][]byte
|
||||
fail error
|
||||
}
|
||||
|
||||
func (w *memWriter) Put(key string, value []byte) error {
|
||||
if w.fail != nil {
|
||||
return w.fail
|
||||
}
|
||||
if w.puts == nil {
|
||||
w.puts = map[string][]byte{}
|
||||
}
|
||||
w.puts[key] = value
|
||||
return nil
|
||||
}
|
||||
|
||||
// holdAsk is a key of the assigner's asks: the ask about a held fingerprint, as a fixture (in release 5 the assigner writes
|
||||
// none yet; the value is `{"v":1,"ask":<the router's ask>}`, as the gate's minus group and heads).
|
||||
func holdAsk(t *testing.T) (Entry, asks.Ask) {
|
||||
t.Helper()
|
||||
key, err := delivery.HoldAskKey("plex", fpNew)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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 := delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: a})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return Entry{Bucket: delivery.AssignerAsksBucket, Key: key, Value: v, At: t0}, a
|
||||
}
|
||||
|
||||
func TestAnAskNeedingNoProofIsOpenInTheViewWithItsOptions(t *testing.T) { // 1297/B9
|
||||
m := New()
|
||||
e, a := holdAsk(t)
|
||||
fill(m, e)
|
||||
got := m.Snapshot().Asks
|
||||
if len(got) != 1 || got[0].Key != e.Key || got[0].Headline != a.Headline || got[0].How != AnswerInTheView || strings.Join(got[0].Options, ",") != "release,keep" {
|
||||
t.Fatalf("asks = %+v", got)
|
||||
}
|
||||
// Once the teller filled it (its answer's warrant names the ask's id) it is answered and leaves the list.
|
||||
fill(m, answerToAsking(t, e.Key, a.ID))
|
||||
if got := m.Snapshot().Asks; len(got) != 0 {
|
||||
t.Errorf("an answered ask is listed: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheViewWritesItsAnswerUnderTheAsksViewKey(t *testing.T) { // 1297/B9a AC
|
||||
m, w := New(), &memWriter{}
|
||||
m.SetWriter(w)
|
||||
e, _ := holdAsk(t)
|
||||
fill(m, e)
|
||||
if err := m.Answer(e.Key, "release", "the check was a flake"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw, ok := w.puts[e.Key+".view"]
|
||||
if !ok || len(w.puts) != 1 {
|
||||
t.Fatalf("puts = %v, want the one key %s.view", w.puts, e.Key)
|
||||
}
|
||||
a, err := delivery.DecodeViewAnswer(raw)
|
||||
if err != nil || a.Option != "release" || a.Words != "the check was a flake" {
|
||||
t.Errorf("answer = %+v, %v", a, err)
|
||||
}
|
||||
// The key is the SDK's: the bucket's key parses back to the ask.
|
||||
if back, err := delivery.ParseViewAnswerKey(e.Key + ".view"); err != nil || back != e.Key {
|
||||
t.Errorf("parse = %q, %v", back, err)
|
||||
}
|
||||
// The view hears its own answer back from its bucket and shows the ask as answered, waiting for the teller.
|
||||
fill(m, Entry{Bucket: delivery.ViewAnswersBucket, Key: e.Key + ".view", Value: raw, At: t0})
|
||||
got := m.Snapshot().Asks
|
||||
if len(got) != 1 || got[0].Chosen != "release" {
|
||||
t.Errorf("asks = %+v, want the ask with chosen release", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheViewRefusesWhatItMayNotAnswer(t *testing.T) { // 1297/B9a AC
|
||||
e, a := holdAsk(t)
|
||||
merge := askEntry(t, "feat/x", "Merge into main: mesh-delivery#12", "Merges.")
|
||||
cases := []struct {
|
||||
name, key, option, want string
|
||||
prep []Entry
|
||||
}{
|
||||
{"the merge's Approve", merge.Key, "approve", "answer it on the phone (issue 953)", []Entry{merge}},
|
||||
{"an option the ask does not offer", e.Key, "delete", `offers no option "delete"`, []Entry{e}},
|
||||
{"an ask the view does not hold", e.Key, "release", "no open ask", nil},
|
||||
{"an ask that is no ask key", "hold.x", "release", "not an ask key", nil},
|
||||
{"an ask the teller has filled", e.Key, "release", "no open ask", []Entry{e, answerToAsking(t, e.Key, a.ID)}},
|
||||
{"no option", e.Key, " ", "no option", []Entry{e}},
|
||||
}
|
||||
for _, c := range cases {
|
||||
m, w := New(), &memWriter{}
|
||||
m.SetWriter(w)
|
||||
fill(m, c.prep...)
|
||||
err := m.Answer(c.key, c.option, "")
|
||||
if err == nil || !strings.Contains(err.Error(), c.want) {
|
||||
t.Errorf("%s: err = %v, want it to say %q", c.name, err, c.want)
|
||||
}
|
||||
if len(w.puts) != 0 {
|
||||
t.Errorf("%s: wrote %v", c.name, w.puts)
|
||||
}
|
||||
}
|
||||
m := New()
|
||||
e2, _ := holdAsk(t)
|
||||
fill(m, e2)
|
||||
if err := m.Answer(e2.Key, "release", ""); err == nil || !strings.Contains(err.Error(), "cannot write") {
|
||||
t.Errorf("a view with no writer: %v", err)
|
||||
}
|
||||
m.SetWriter(&memWriter{fail: fmt.Errorf("the bus said no")})
|
||||
if err := m.Answer(e2.Key, "release", ""); err == nil || !strings.Contains(err.Error(), "the bus said no") {
|
||||
t.Errorf("a failed write: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheSocketAnswersAnAsk(t *testing.T) { // 1297/B9b
|
||||
path, m := served(t)
|
||||
w := &memWriter{}
|
||||
m.SetWriter(w)
|
||||
e, _ := holdAsk(t)
|
||||
fill(m, e)
|
||||
if err := AnswerAsk(path, e.Key, "keep", "wait for the next build"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := w.puts[e.Key+".view"]; !ok {
|
||||
t.Fatalf("puts = %v", w.puts)
|
||||
}
|
||||
merge := askEntry(t, "feat/x", "Merge", "Merges.")
|
||||
fill(m, merge)
|
||||
err := AnswerAsk(path, merge.Key, "approve", "")
|
||||
if err == nil || !strings.Contains(err.Error(), "answer it on the phone (issue 953)") {
|
||||
t.Errorf("merge over the socket: %v", err)
|
||||
}
|
||||
// the old asks stay refused
|
||||
c, err := dial(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
c.Write([]byte(`{"ask":"answer"}` + "\n"))
|
||||
line, _ := bufio.NewReader(c).ReadString('\n')
|
||||
c.Close()
|
||||
if !strings.Contains(line, `"error"`) {
|
||||
t.Errorf("an answer naming nothing: %q", line)
|
||||
}
|
||||
}
|
||||
|
||||
func postAnswer(h http.Handler, form url.Values, mod func(*http.Request)) *httptest.ResponseRecorder {
|
||||
req := httptest.NewRequest(http.MethodPost, "/answer", strings.NewReader(form.Encode()))
|
||||
req.Host = "127.0.0.1:8471"
|
||||
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
|
||||
if mod != nil {
|
||||
mod(req)
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, req)
|
||||
return rec
|
||||
}
|
||||
|
||||
func TestThePageAnswersAnAskAndRefusesAForeignPost(t *testing.T) { // 1297/B9a
|
||||
m, w := New(), &memWriter{}
|
||||
m.SetWriter(w)
|
||||
e, _ := holdAsk(t)
|
||||
fill(m, e)
|
||||
h := Web(m)
|
||||
page := httptest.NewRecorder()
|
||||
h.ServeHTTP(page, httptest.NewRequest(http.MethodGet, "/", nil))
|
||||
for _, want := range []string{`action="/answer"`, `value="release"`, `value="keep"`, e.Key} {
|
||||
if !strings.Contains(page.Body.String(), want) {
|
||||
t.Errorf("the page lacks %q:\n%s", want, page.Body.String())
|
||||
}
|
||||
}
|
||||
form := url.Values{"key": {e.Key}, "option": {"release"}, "words": {"ok"}}
|
||||
// A post from another origin, or to a host name that is not the loopback's, is refused and writes nothing.
|
||||
for name, mod := range map[string]func(*http.Request){
|
||||
"another origin": func(r *http.Request) { r.Header.Set("Origin", "http://evil.example") },
|
||||
"another host": func(r *http.Request) { r.Host = "evil.example" },
|
||||
} {
|
||||
if rec := postAnswer(h, form, mod); rec.Code != http.StatusForbidden || len(w.puts) != 0 {
|
||||
t.Errorf("%s: %d, puts %v", name, rec.Code, w.puts)
|
||||
}
|
||||
}
|
||||
rec := postAnswer(h, form, func(r *http.Request) { r.Header.Set("Origin", "http://127.0.0.1:8471") })
|
||||
if rec.Code != http.StatusSeeOther || len(w.puts) != 1 {
|
||||
t.Fatalf("answer = %d %q, puts %v", rec.Code, rec.Body.String(), w.puts)
|
||||
}
|
||||
// The merge's Approve is refused with its reason, and the rest of the page stays read-only.
|
||||
merge := askEntry(t, "feat/x", "Merge", "Merges.")
|
||||
fill(m, merge)
|
||||
rec = postAnswer(h, url.Values{"key": {merge.Key}, "option": {"approve"}}, nil)
|
||||
if rec.Code != http.StatusUnprocessableEntity || !strings.Contains(rec.Body.String(), "answer it on the phone (issue 953)") || len(w.puts) != 1 {
|
||||
t.Errorf("merge = %d %q", rec.Code, rec.Body.String())
|
||||
}
|
||||
if rec := httptest.NewRecorder(); true {
|
||||
h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/answer", nil))
|
||||
if rec.Code != http.StatusMethodNotAllowed {
|
||||
t.Errorf("GET /answer = %d", rec.Code)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
package view
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
const (
|
||||
fpOld = "src2:1111111111111111111111111111111111111111111111111111111111111111"
|
||||
fpNew = "src2:2222222222222222222222222222222222222222222222222222222222222222"
|
||||
)
|
||||
|
||||
// diffEntry is a key of the assigner's differences as the assigner writes it (mesh-sdk go/delivery EncodeDifference).
|
||||
func differenceE(t *testing.T, node, module string, d delivery.Difference) Entry {
|
||||
t.Helper()
|
||||
key, err := delivery.DifferenceKey(node, module)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
v, err := delivery.EncodeDifference(d)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return Entry{Bucket: delivery.AssignerDifferencesBucket, Key: key, Value: v, At: t0}
|
||||
}
|
||||
|
||||
func TestEveryDifferenceIsListedWithItsRule(t *testing.T) { // 1297/B8 AC
|
||||
m := New()
|
||||
fill(m,
|
||||
differenceE(t, "ace", "plex", delivery.Difference{Want: fpOld, Sent: fpNew, Rule: delivery.RuleCanary, Words: "ace is not the canary"}),
|
||||
differenceE(t, "g14", "plex", delivery.Difference{Want: fpOld, Sent: fpNew, Unexplained: true}),
|
||||
differenceE(t, "ace", "old", delivery.Difference{WantRemoved: true, Sent: fpOld, Rule: delivery.RuleNotHeld}),
|
||||
)
|
||||
got := m.Snapshot().Differences
|
||||
if len(got) != 3 {
|
||||
t.Fatalf("differences = %+v, want all three", got)
|
||||
}
|
||||
// ordered by node then module
|
||||
if got[0].Node != "ace" || got[0].Module != "old" || got[1].Module != "plex" || got[2].Node != "g14" {
|
||||
t.Errorf("order = %+v", got)
|
||||
}
|
||||
if r := got[1]; r.Want != fpOld || r.Sent != fpNew || r.Rule != delivery.RuleCanary || r.Unexplained || r.Words != "ace is not the canary" {
|
||||
t.Errorf("row = %+v", r)
|
||||
}
|
||||
if !got[0].WantRemoved || got[0].Rule != delivery.RuleNotHeld {
|
||||
t.Errorf("removed row = %+v", got[0])
|
||||
}
|
||||
if !got[2].Unexplained || got[2].Rule != 0 {
|
||||
t.Errorf("unexplained row = %+v", got[2])
|
||||
}
|
||||
// a difference taken away (the want and the send agree again) leaves the list
|
||||
fill(m, Entry{Bucket: delivery.AssignerDifferencesBucket, Key: "g14.plex", Deleted: true})
|
||||
if got := m.Snapshot().Differences; len(got) != 2 {
|
||||
t.Errorf("after the delete = %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnUnreadableDifferenceIsNamedAndNotDropped(t *testing.T) { // never dropped silently
|
||||
m := New()
|
||||
fill(m, Entry{Bucket: delivery.AssignerDifferencesBucket, Key: "ace.plex", Value: []byte(`{"v":99}`), At: t0})
|
||||
s := m.Snapshot()
|
||||
if len(s.Differences) != 0 || len(s.Unreadable) != 1 || s.Unreadable[0].Bucket != delivery.AssignerDifferencesBucket || s.Unreadable[0].Key != "ace.plex" {
|
||||
t.Errorf("snapshot = %+v", s)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTableAndPageShowTheWantsAgainstTheSends(t *testing.T) { // 1297/B8 AC
|
||||
m := New()
|
||||
fill(m,
|
||||
differenceE(t, "ace", "plex", delivery.Difference{Want: fpOld, Sent: fpNew, Rule: delivery.RuleCanary, Words: "ace is not the canary"}),
|
||||
differenceE(t, "g14", "plex", delivery.Difference{Want: fpOld, Sent: fpNew, Unexplained: true}),
|
||||
)
|
||||
var out strings.Builder
|
||||
Table(&out, m.Snapshot())
|
||||
for _, w := range []string{"WANTS AGAINST SENDS", "ace", "plex", fpOld, fpNew, "rule 3", "ace is not the canary", "unexplained"} {
|
||||
if !strings.Contains(out.String(), w) {
|
||||
t.Errorf("table lacks %q:\n%s", w, out.String())
|
||||
}
|
||||
}
|
||||
for _, path := range []string{"/", "/wants"} {
|
||||
rec := httptest.NewRecorder()
|
||||
Web(m).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, path, nil))
|
||||
body := rec.Body.String()
|
||||
if rec.Code != 200 {
|
||||
t.Fatalf("%s = %d", path, rec.Code)
|
||||
}
|
||||
for _, w := range []string{"WANTS AGAINST SENDS", "rule 3", "unexplained", "ace is not the canary"} {
|
||||
if !strings.Contains(body, w) {
|
||||
t.Errorf("%s lacks %q:\n%s", path, w, body)
|
||||
}
|
||||
}
|
||||
}
|
||||
// With none, the page says so and does not hide the heading.
|
||||
rec := httptest.NewRecorder()
|
||||
Web(New()).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/wants", nil))
|
||||
if !strings.Contains(rec.Body.String(), "no difference") {
|
||||
t.Errorf("empty page = %q", rec.Body.String())
|
||||
}
|
||||
}
|
||||
@@ -42,10 +42,14 @@ func ListenSocket(path string) (net.Listener, error) {
|
||||
}
|
||||
|
||||
type ask struct {
|
||||
Ask string `json:"ask"`
|
||||
Ask string `json:"ask"`
|
||||
Key string `json:"key,omitempty"`
|
||||
Option string `json:"option,omitempty"`
|
||||
Words string `json:"words,omitempty"`
|
||||
}
|
||||
|
||||
// ServeSocket answers {"ask":"rows"} on l, line-delimited JSON, until ctx ends. Any other ask is refused.
|
||||
// ServeSocket answers {"ask":"rows"} on l, line-delimited JSON, until ctx ends, and {"ask":"answer","key":…,"option":…,"words":…}
|
||||
// with the model's Answer (the one thing the view writes, issue 1297 B9). Any other ask is refused.
|
||||
func ServeSocket(ctx context.Context, l net.Listener, m *Model) error {
|
||||
go func() { <-ctx.Done(); l.Close() }()
|
||||
for {
|
||||
@@ -68,13 +72,46 @@ func answer(c net.Conn, m *Model) {
|
||||
return
|
||||
}
|
||||
var a ask
|
||||
if json.Unmarshal(line, &a) != nil || a.Ask != "rows" {
|
||||
json.NewEncoder(c).Encode(map[string]string{"error": `refused: this socket answers {"ask":"rows"} and nothing else`})
|
||||
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`})
|
||||
return
|
||||
}
|
||||
if a.Ask == "answer" {
|
||||
if err := m.Answer(a.Key, a.Option, a.Words); err != nil {
|
||||
json.NewEncoder(c).Encode(map[string]string{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
json.NewEncoder(c).Encode(map[string]string{"answered": a.Key})
|
||||
return
|
||||
}
|
||||
json.NewEncoder(c).Encode(m.Snapshot())
|
||||
}
|
||||
|
||||
// AnswerAsk asks the socket at path to write the view's answer to the open ask at key. The view's refusal is the error.
|
||||
func AnswerAsk(path, key, option, words string) 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))
|
||||
req, _ := json.Marshal(ask{Ask: "answer", Key: key, Option: option, Words: words})
|
||||
if _, err := c.Write(append(req, '\n')); err != nil {
|
||||
return err
|
||||
}
|
||||
var reply struct {
|
||||
Answered string `json:"answered"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
if err := json.NewDecoder(c).Decode(&reply); err != nil {
|
||||
return fmt.Errorf("the view's reply is not an answer: %w", err)
|
||||
}
|
||||
if reply.Error != "" {
|
||||
return errors.New(reply.Error)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Ask asks the socket at path for the rows.
|
||||
func Ask(path string) (Snapshot, error) {
|
||||
c, err := net.DialTimeout("unix", path, 5*time.Second)
|
||||
@@ -160,11 +197,47 @@ func fingerprintText(s Snapshot) [][]string {
|
||||
func askText(s Snapshot) [][]string {
|
||||
var out [][]string
|
||||
for _, a := range s.Asks {
|
||||
out = append(out, []string{a.Key, a.Headline, strings.Join(strings.Fields(strings.ReplaceAll(a.Words, "\n", " / ")), " "), strings.Join(a.Pulls, ", "), a.How})
|
||||
how := a.How
|
||||
switch {
|
||||
case a.Chosen != "":
|
||||
how = "answered " + a.Chosen + " in the view; the teller fills it"
|
||||
case len(a.Options) > 0:
|
||||
how += ": " + strings.Join(a.Options, " or ")
|
||||
}
|
||||
out = append(out, []string{a.Key, a.Headline, strings.Join(strings.Fields(strings.ReplaceAll(a.Words, "\n", " / ")), " "), strings.Join(a.Pulls, ", "), how})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// differenceText is one line per difference: the node, the module, what the assigner wants, what the controller sends, the rule
|
||||
// that explains it (or unexplained) and the assigner's words.
|
||||
func differenceText(s Snapshot) [][]string {
|
||||
var out [][]string
|
||||
for _, d := range s.Differences {
|
||||
want, sent := d.Want, d.Sent
|
||||
switch {
|
||||
case d.WantRemoved:
|
||||
want = "removed"
|
||||
case want == "":
|
||||
want = "none"
|
||||
}
|
||||
if sent == "" {
|
||||
sent = "none"
|
||||
if d.SentBuild != "" {
|
||||
sent = "build " + d.SentBuild
|
||||
}
|
||||
}
|
||||
why := "unexplained"
|
||||
if !d.Unexplained {
|
||||
why = fmt.Sprintf("rule %d", d.Rule)
|
||||
}
|
||||
out = append(out, []string{d.Node, d.Module, want, sent, why, d.Words})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
var differenceHead = []string{"NODE", "MODULE", "WANT", "SENT", "RULE", "WORDS"}
|
||||
|
||||
var head = []string{"NODE", "MODULE", "ASSIGNED", "HAVE", "PHASE", "HEALTH", "SINCE", "DETAIL", "WAITS"}
|
||||
|
||||
// Table prints the snapshot as a table.
|
||||
@@ -204,6 +277,14 @@ func Table(w io.Writer, s Snapshot) {
|
||||
row(cells)
|
||||
}
|
||||
}
|
||||
if d := differenceText(s); len(d) > 0 {
|
||||
fmt.Fprintln(tw)
|
||||
row([]string{"WANTS AGAINST SENDS"})
|
||||
row(differenceHead)
|
||||
for _, cells := range d {
|
||||
row(cells)
|
||||
}
|
||||
}
|
||||
if f := fingerprintText(s); len(f) > 0 {
|
||||
fmt.Fprintln(tw)
|
||||
row([]string{"MODULE", "FINGERPRINT", "INDEXER AGAINST BUILD"})
|
||||
@@ -247,20 +328,41 @@ var page = template.Must(template.New("p").Parse(`<!doctype html><html><head><me
|
||||
<body><h1>What each node runs</h1><table><tr>{{range .Head}}<th>{{.}}</th>{{end}}</tr>
|
||||
{{range .Rows}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>
|
||||
{{if .Groups}}<h2>CANDIDATES</h2><table><tr><th>GROUP</th><th>PULL REQUEST</th><th>HEAD</th><th>MODULES</th></tr>{{range .Groups}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
|
||||
{{if .Asks}}<h2>ASKS</h2><table><tr><th>ASK</th><th>HEADLINE</th><th>WORDS</th><th>PULL REQUESTS</th><th>ANSWER</th></tr>{{range .Asks}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
|
||||
{{if .Asks}}<h2>ASKS</h2><table><tr><th>ASK</th><th>HEADLINE</th><th>WORDS</th><th>PULL REQUESTS</th><th>ANSWER</th></tr>{{range .Asks}}<tr>{{range .Cells}}<td>{{.}}</td>{{end}}<td>{{if .Options}}<form method="post" action="/answer"><input type="hidden" name="key" value="{{.Key}}"><input name="words" placeholder="words (optional)">{{range .Options}} <button name="option" value="{{.}}">{{.}}</button>{{end}}</form>{{end}}</td></tr>{{end}}</table>{{end}}
|
||||
{{if .Differences}}<h2>WANTS AGAINST SENDS</h2><table><tr>{{range .DifferenceHead}}<th>{{.}}</th>{{end}}</tr>{{range .Differences}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
|
||||
{{if .Fingerprints}}<h2>FINGERPRINTS</h2><table><tr><th>MODULE</th><th>STATE</th><th>INDEXER AGAINST BUILD</th></tr>{{range .Fingerprints}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
|
||||
{{if .Offline}}<h2>OFFLINE</h2><table>{{range .Offline}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
|
||||
{{if .Stuck}}<h2>STUCK</h2><table><tr><th>BUCKET</th><th>KEY</th><th>WAITED FOR</th><th>SINCE</th></tr>{{range .Stuck}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
|
||||
{{if .Unreadable}}<h2>UNREADABLE</h2><table>{{range .Unreadable}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}</body></html>`))
|
||||
|
||||
var wantsPage = template.Must(template.New("w").Parse(`<!doctype html><html><head><meta charset="utf-8"><meta http-equiv="refresh" content="5">
|
||||
<title>Wants against sends</title><style>body{font:14px sans-serif;margin:1em}table{border-collapse:collapse}td,th{padding:2px 10px;text-align:left}</style></head>
|
||||
<body><p><a href="/">What each node runs</a></p><h1>WANTS AGAINST SENDS</h1>
|
||||
{{if .Differences}}<table><tr>{{range .Head}}<th>{{.}}</th>{{end}}</tr>{{range .Differences}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>
|
||||
{{else}}<p>no difference: every want is what the controller sends.</p>{{end}}</body></html>`))
|
||||
|
||||
// Web is the read-only web view of the same rows, refreshed by the browser.
|
||||
func Web(m *Model) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path == "/answer" && r.Method == http.MethodPost {
|
||||
answerForm(m, w, r)
|
||||
return
|
||||
}
|
||||
if r.URL.Path == "/answer" {
|
||||
w.Header().Set("Allow", "POST")
|
||||
http.Error(w, "an answer is posted from the form of the page", http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
if r.Method != http.MethodGet && r.Method != http.MethodHead {
|
||||
w.Header().Set("Allow", "GET, HEAD")
|
||||
http.Error(w, "this view is read-only", http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
if r.URL.Path == "/wants" {
|
||||
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
||||
wantsPage.Execute(w, map[string]any{"Head": differenceHead, "Differences": differenceText(m.Snapshot())})
|
||||
return
|
||||
}
|
||||
if r.URL.Path != "/" {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
@@ -281,6 +383,62 @@ func Web(m *Model) http.Handler {
|
||||
for _, u := range snap.Unreadable {
|
||||
bad = append(bad, []string{u.Bucket, u.Key, u.Error})
|
||||
}
|
||||
page.Execute(w, map[string]any{"Head": head, "Rows": rows, "Groups": groupText(snap), "Asks": askText(snap), "Fingerprints": fingerprintText(snap), "Offline": off, "Stuck": stuck, "Unreadable": bad})
|
||||
page.Execute(w, map[string]any{"Head": head, "Rows": rows, "Groups": groupText(snap), "Asks": askPage(snap), "Fingerprints": fingerprintText(snap), "Differences": differenceText(snap), "DifferenceHead": differenceHead, "Offline": off, "Stuck": stuck, "Unreadable": bad})
|
||||
})
|
||||
}
|
||||
|
||||
// askPageRow is an open ask on the page: its cells, and the options a form offers when the view may answer it.
|
||||
type askPageRow struct {
|
||||
Cells []string
|
||||
Key string
|
||||
Options []string
|
||||
}
|
||||
|
||||
func askPage(s Snapshot) []askPageRow {
|
||||
var out []askPageRow
|
||||
cells := askText(s)
|
||||
for i, a := range s.Asks {
|
||||
row := askPageRow{Cells: cells[i], Key: a.Key}
|
||||
if a.Chosen == "" {
|
||||
row.Options = a.Options
|
||||
}
|
||||
out = append(out, row)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// loopback says whether a Host header names the machine itself, as the web view listens on 127.0.0.1 only: a page of another
|
||||
// origin, or a name an attacker's DNS points here, writes nothing.
|
||||
func loopback(host string) bool {
|
||||
h, _, err := net.SplitHostPort(host)
|
||||
if err != nil {
|
||||
h = host
|
||||
}
|
||||
return h == "127.0.0.1" || h == "localhost" || h == "[::1]" || h == "::1"
|
||||
}
|
||||
|
||||
// answerForm is the one write the web view makes: POST /answer with key, option and words, from a form of this page.
|
||||
func answerForm(m *Model, w http.ResponseWriter, r *http.Request) {
|
||||
if !loopback(r.Host) {
|
||||
http.Error(w, "this view answers only on its own address", http.StatusForbidden)
|
||||
return
|
||||
}
|
||||
if o := r.Header.Get("Origin"); o != "" && o != "http://"+r.Host {
|
||||
http.Error(w, "this view answers only a form of its own page", http.StatusForbidden)
|
||||
return
|
||||
}
|
||||
if r.Header.Get("Sec-Fetch-Site") == "cross-site" {
|
||||
http.Error(w, "this view answers only a form of its own page", http.StatusForbidden)
|
||||
return
|
||||
}
|
||||
r.Body = http.MaxBytesReader(w, r.Body, 8192)
|
||||
if err := r.ParseForm(); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if err := m.Answer(r.PostFormValue("key"), r.PostFormValue("option"), r.PostFormValue("words")); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusUnprocessableEntity)
|
||||
return
|
||||
}
|
||||
http.Redirect(w, r, "/", http.StatusSeeOther)
|
||||
}
|
||||
|
||||
@@ -4,11 +4,13 @@ package view
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
@@ -64,6 +66,24 @@ type Snapshot struct {
|
||||
Fingerprints []FingerprintRow `json:"fingerprints"`
|
||||
// Asks is every ask the gate holds that has no answer yet: what waits for the operator's Approve.
|
||||
Asks []AskRow `json:"asks"`
|
||||
// Differences is every difference between what the assigner wants a node to run and what the controller sends, with the rule that
|
||||
// explains it or unexplained (novox/hq ADR 0528 clause 4, issue 1297 B8).
|
||||
Differences []DifferenceRow `json:"differences"`
|
||||
}
|
||||
|
||||
// DifferenceRow is a module on a node whose want and send differ: the fingerprint wanted (empty when the node has no want,
|
||||
// WantRemoved when the want is removed), the fingerprint of the build the controller's assignment names (empty when it sends
|
||||
// none; SentBuild names that build), and the rule that explains it (Rule 1 to 6) or Unexplained.
|
||||
type DifferenceRow struct {
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
Want string `json:"want,omitempty"`
|
||||
WantRemoved bool `json:"want-removed,omitempty"`
|
||||
Sent string `json:"sent,omitempty"`
|
||||
SentBuild string `json:"sent-build,omitempty"`
|
||||
Rule int `json:"rule,omitempty"`
|
||||
Unexplained bool `json:"unexplained,omitempty"`
|
||||
Words string `json:"words,omitempty"`
|
||||
}
|
||||
|
||||
// AskRow is an open ask (novox/hq ADR 0437 §10): the gate's ask with no answer under its key in the teller's answers.
|
||||
@@ -75,12 +95,19 @@ type AskRow struct {
|
||||
Pulls []string `json:"pulls"`
|
||||
Expires time.Time `json:"expires,omitzero"`
|
||||
How string `json:"how"`
|
||||
// Options is what the view offers to answer with, for an ask that needs no proof of who answered (the assigner's hold and
|
||||
// restore, ADR 0528 clause 6); Chosen is the option the view wrote that the teller has not filled yet.
|
||||
Options []string `json:"options,omitempty"`
|
||||
Chosen string `json:"chosen,omitempty"`
|
||||
}
|
||||
|
||||
// AnswerOnThePhone is what an open ask says about answering it: the view only shows, and the Approve is answered on the
|
||||
// operator's phone (ADR 0437 §10: answering in the view is release 5's).
|
||||
const AnswerOnThePhone = "answer on the phone"
|
||||
|
||||
// AnswerInTheView is what an open ask that needs no proof of who answered says (ADR 0528 clause 6): the operator answers it here.
|
||||
const AnswerInTheView = "answer in the view"
|
||||
|
||||
// GroupRow is a candidate group (novox/hq ADR 0437 §5): the branch its pull requests share and the pull requests.
|
||||
type GroupRow struct {
|
||||
Group string `json:"group"`
|
||||
@@ -150,13 +177,17 @@ 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
|
||||
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
|
||||
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
|
||||
haves map[string]haveEntry
|
||||
nodes map[string]nodeEntry
|
||||
// the watchman's buckets, by key
|
||||
@@ -221,6 +252,16 @@ type keyEntry struct {
|
||||
err string
|
||||
}
|
||||
|
||||
type diffEntry struct {
|
||||
value delivery.Difference
|
||||
err string
|
||||
}
|
||||
|
||||
type assignerAskRecord struct {
|
||||
ask asks.Ask
|
||||
err string
|
||||
}
|
||||
|
||||
type moduleEntry struct {
|
||||
value delivery.Module
|
||||
err string
|
||||
@@ -245,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{}, 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{}}
|
||||
}
|
||||
@@ -321,6 +362,32 @@ func (m *Model) Apply(e Entry) {
|
||||
}
|
||||
a, err := delivery.DecodeAsk(e.Value)
|
||||
m.asks[e.Key] = askRecord{value: a, err: errText(err)}
|
||||
case delivery.AssignerAsksBucket:
|
||||
if e.Deleted {
|
||||
delete(m.aasks, e.Key)
|
||||
return
|
||||
}
|
||||
a, err := decodeAssignerAsk(e.Key, e.Value)
|
||||
m.aasks[e.Key] = assignerAskRecord{ask: a, err: errText(err)}
|
||||
case delivery.ViewAnswersBucket:
|
||||
askKey, err := delivery.ParseViewAnswerKey(e.Key)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if e.Deleted {
|
||||
delete(m.chosen, askKey)
|
||||
return
|
||||
}
|
||||
if a, err := delivery.DecodeViewAnswer(e.Value); err == nil {
|
||||
m.chosen[askKey] = a
|
||||
}
|
||||
case delivery.AssignerDifferencesBucket:
|
||||
if e.Deleted {
|
||||
delete(m.diffs, e.Key)
|
||||
return
|
||||
}
|
||||
d, err := delivery.DecodeDifference(e.Value)
|
||||
m.diffs[e.Key] = diffEntry{value: d, err: errText(err)}
|
||||
case delivery.AnswersBucket:
|
||||
if e.Deleted {
|
||||
delete(m.answers, e.Key)
|
||||
@@ -490,6 +557,16 @@ func (m *Model) Snapshot() Snapshot {
|
||||
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.AsksBucket, Key: key, Error: a.err})
|
||||
}
|
||||
}
|
||||
for key, a := range m.aasks {
|
||||
if a.err != "" {
|
||||
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.AssignerAsksBucket, Key: key, Error: a.err})
|
||||
}
|
||||
}
|
||||
for key, d := range m.diffs {
|
||||
if d.err != "" {
|
||||
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.AssignerDifferencesBucket, Key: key, Error: d.err})
|
||||
}
|
||||
}
|
||||
sort.Slice(offline, func(i, j int) bool {
|
||||
a, b := offline[i], offline[j]
|
||||
return a.Kind+a.Node+a.Part+a.Instance < b.Kind+b.Node+b.Part+b.Instance
|
||||
@@ -501,7 +578,7 @@ func (m *Model) Snapshot() Snapshot {
|
||||
return unreadable[i].Bucket+"."+unreadable[i].Key < unreadable[j].Bucket+"."+unreadable[j].Key
|
||||
})
|
||||
return Snapshot{Rows: rows, Nodes: nodes, Offline: offline, Stuck: stuck, Unreadable: unreadable,
|
||||
Groups: m.groupRows(), Fingerprints: m.fingerprintRows(), Asks: m.askRows()}
|
||||
Groups: m.groupRows(), Fingerprints: m.fingerprintRows(), Asks: m.askRows(), Differences: m.differenceRows()}
|
||||
}
|
||||
|
||||
// groupRows are the readable candidate groups, ordered by group, their pull requests by repository and number.
|
||||
@@ -655,6 +732,111 @@ func (m *Model) askRows() []AskRow {
|
||||
}
|
||||
out = append(out, row)
|
||||
}
|
||||
for key, a := range m.aasks {
|
||||
if a.err != "" || m.answers[key] == a.ask.ID {
|
||||
continue
|
||||
}
|
||||
row := AskRow{Key: key, Headline: a.ask.Headline, Words: a.ask.Explanation, Expires: a.ask.Expires, How: AnswerInTheView, Pulls: []string{},
|
||||
Chosen: m.chosenOf(key, a.ask)}
|
||||
for _, o := range a.ask.Options {
|
||||
row.Options = append(row.Options, o.ID)
|
||||
}
|
||||
out = append(out, row)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
|
||||
return out
|
||||
}
|
||||
|
||||
// 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) {
|
||||
a, err := delivery.DecodeAssignerAsk(key, value)
|
||||
if err != nil {
|
||||
return asks.Ask{}, fmt.Errorf("the assigner's ask %s is not readable: %w", key, err)
|
||||
}
|
||||
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 ""
|
||||
}
|
||||
|
||||
// Writer puts a key of the view's bucket answers, the only state the view writes.
|
||||
type Writer interface {
|
||||
Put(key string, value []byte) error
|
||||
}
|
||||
|
||||
// SetWriter gives the model the bucket its answers are written to.
|
||||
func (m *Model) SetWriter(w Writer) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.writer = w
|
||||
}
|
||||
|
||||
// Answer writes the view's answer to the open ask at askKey: option, one the ask offers, and the operator's words. It writes
|
||||
// `<ask key>.view` and nothing else; an ask that needs proof of who answered (the merge's Approve, issue 953) is refused with
|
||||
// its reason, as is an ask the view does not hold open, an option the ask does not offer, and one the view answered already.
|
||||
func (m *Model) Answer(askKey, option, words string) error {
|
||||
key, err := delivery.ViewAnswerKey(askKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
option = strings.TrimSpace(option)
|
||||
if option == "" {
|
||||
return fmt.Errorf("an answer names no option")
|
||||
}
|
||||
m.mu.Lock()
|
||||
a, held := m.aasks[askKey]
|
||||
open := held && a.err == "" && m.answers[askKey] != a.ask.ID
|
||||
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)
|
||||
}
|
||||
if !slices.ContainsFunc(a.ask.Options, func(o asks.Option) bool { return o.ID == option }) {
|
||||
ids := []string{}
|
||||
for _, o := range a.ask.Options {
|
||||
ids = append(ids, o.ID)
|
||||
}
|
||||
return fmt.Errorf("the ask %s offers no option %q (it offers %s)", askKey, option, strings.Join(ids, ", "))
|
||||
}
|
||||
if chosen != "" {
|
||||
return fmt.Errorf("the view answered %s with %q already: the teller fills the ask from it", askKey, chosen)
|
||||
}
|
||||
if w == nil {
|
||||
return fmt.Errorf("the view cannot write its answers here")
|
||||
}
|
||||
body, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Asking: a.ask.ID, Option: option, Words: strings.TrimSpace(words)})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return w.Put(key, body)
|
||||
}
|
||||
|
||||
// differenceRows are the readable differences, ordered by node and module.
|
||||
func (m *Model) differenceRows() []DifferenceRow {
|
||||
out := []DifferenceRow{}
|
||||
for key, d := range m.diffs {
|
||||
if d.err != "" {
|
||||
continue
|
||||
}
|
||||
node, module, err := delivery.ParseDifferenceKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
v := d.value
|
||||
out = append(out, DifferenceRow{Node: node, Module: module, Want: v.Want, WantRemoved: v.WantRemoved, Sent: v.Sent, SentBuild: v.SentBuild,
|
||||
Rule: v.Rule, Unexplained: v.Unexplained, Words: v.Words})
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool {
|
||||
if out[i].Node != out[j].Node {
|
||||
return out[i].Node < out[j].Node
|
||||
}
|
||||
return out[i].Module < out[j].Module
|
||||
})
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -36,7 +36,13 @@ type item struct {
|
||||
Template string `json:"template"`
|
||||
}
|
||||
|
||||
type state struct {
|
||||
Name string `json:"name"`
|
||||
History int `json:"history,omitempty"`
|
||||
}
|
||||
|
||||
type manifest struct {
|
||||
State []state `json:"state"`
|
||||
Module string `json:"module"`
|
||||
Moved bool `json:"moved"`
|
||||
Version string `json:"version"`
|
||||
@@ -76,17 +82,21 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
|
||||
if m.Module != "mesh-delivery-view" {
|
||||
t.Errorf("module = %q", m.Module)
|
||||
}
|
||||
want := []string{"mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-builder.builds", "mesh-delivery-gate.asks", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // 1168/AC1, 1241/AC4, 1345/AC3
|
||||
want := []string{"mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-builder.builds", "mesh-delivery-gate.asks", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
|
||||
got := slices.Clone(m.Reads)
|
||||
slices.Sort(got)
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("reads = %v, want exactly %v", got, want)
|
||||
}
|
||||
for _, k := range []string{"state", "emits", "consumes", "provides", "claims"} {
|
||||
for _, k := range []string{"emits", "consumes", "provides", "claims"} {
|
||||
if _, ok := top[k]; ok {
|
||||
t.Errorf("the manifest declares %q: the view writes nothing", k)
|
||||
t.Errorf("the manifest declares %q: the view writes only its answers", k)
|
||||
}
|
||||
}
|
||||
// The one state the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the teller reads.
|
||||
if want := []state{{Name: "answers", History: 5}}; !reflect.DeepEqual(m.State, want) {
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
|
||||
t.Errorf("own-secrets = %v, want exactly broker (or no credential is issued)", m.OwnSecrets)
|
||||
}
|
||||
|
||||
@@ -11,9 +11,14 @@
|
||||
"builder_build",
|
||||
"builder_builds"
|
||||
],
|
||||
"state": [
|
||||
{ "name": "answers", "history": 5 }
|
||||
],
|
||||
"reads": [
|
||||
"mesh-controller.assignments",
|
||||
"mesh-controller.builds",
|
||||
"mesh-delivery-assigner.asks",
|
||||
"mesh-delivery-assigner.differences",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-gate.asks",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
|
||||
Reference in New Issue
Block a user