Let the view answer the assigner's asks that need no proof, under the ask's view key, and refuse the merge's Approve on the phone's behalf (issue 1297 B9a)
This commit is contained in:
@@ -8,6 +8,7 @@ import (
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -472,3 +473,132 @@ func TestProcessWholeListsEveryDifference(t *testing.T) { // 1297/B8 AC
|
||||
}
|
||||
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, _ := json.Marshal(map[string]any{"v": 1, "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, delivery.AssignerDifferencesBucket} {
|
||||
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) {
|
||||
|
||||
Reference in New Issue
Block a user