jschoubben
6b63f13da6
Seven modules stayed held on every machine and the operator was never asked: the teller only filled assigner asks from the view, a single failed check held a fingerprint for good, and the held record kept no cause. The teller now asks every assigner ask as it asks merge asks and writes the phone's answer under the ask's key; the assigner lifts a hold whose fingerprint runs healthy on every node that runs the module, says so once and takes its ask back; the held record names health, phase, when and the node's words.
644 lines
25 KiB
Go
644 lines
25 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats-server/v2/server"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
|
|
// controller's user and the teller's own principal <node>.<module>.
|
|
const busConf = `
|
|
listen: 127.0.0.1:-1
|
|
jetstream { store_dir: %q }
|
|
accounts {
|
|
MESH {
|
|
jetstream: enabled
|
|
users = [
|
|
{ user: "ctl", password: "pw" }
|
|
{ user: "n1.mesh-delivery-teller", password: "pw" }
|
|
]
|
|
}
|
|
}
|
|
`
|
|
|
|
type syncBuffer struct {
|
|
mu sync.Mutex
|
|
b strings.Builder
|
|
}
|
|
|
|
func (s *syncBuffer) Write(p []byte) (int, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return s.b.Write(p)
|
|
}
|
|
func (s *syncBuffer) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() }
|
|
|
|
// TestTellerWaitsForTheGate starts the program the way the node does (its entry, run, with MESH_BROKER_FILE) against an
|
|
// embedded bus where mesh-delivery-gate_asks does not exist: it keeps running until stopped, says once that it waits for
|
|
// the gate's asks, and writes and publishes nothing (the revised 1240/AC2: the watch is release 3d's).
|
|
func TestTellerWaitsForTheGate(t *testing.T) { // 1240/AC2
|
|
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "t")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { os.RemoveAll(dir) })
|
|
conf := filepath.Join(dir, "bus.conf")
|
|
if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
opts, err := server.ProcessConfigFile(conf)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
opts.NoLog, opts.NoSigs = true, true
|
|
srv, err := server.NewServer(opts)
|
|
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)
|
|
|
|
ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(ctl.Close)
|
|
js, err := jetstream.New(ctl)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
broker := filepath.Join(t.TempDir(), "broker")
|
|
cred, _ := json.Marshal(credential{URL: 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(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) })
|
|
|
|
for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline) && !strings.Contains(logs.String(), "waiting for the gate's asks"); time.Sleep(20 * time.Millisecond) {
|
|
}
|
|
if n := strings.Count(logs.String(), "waiting for the gate's asks"); n != 1 {
|
|
t.Fatalf("the teller said it waits %d times, want once: %q", n, logs.String())
|
|
}
|
|
select {
|
|
case err := <-rn.done:
|
|
t.Fatalf("the teller stopped: %v", err)
|
|
case <-time.After(500 * time.Millisecond):
|
|
}
|
|
if n := strings.Count(logs.String(), "waiting for the gate's asks"); n != 1 {
|
|
t.Errorf("the teller said it waits %d times after a while, want once", n)
|
|
}
|
|
// It writes and publishes nothing: no stream, no bucket exists.
|
|
names := js.StreamNames(ctx)
|
|
for name := range names.Name() {
|
|
t.Errorf("a stream %s exists: the teller wrote something", name)
|
|
}
|
|
if _, err := js.KeyValue(ctx, "mesh-delivery-gate_asks"); err == nil {
|
|
t.Error("the gate's asks bucket exists")
|
|
}
|
|
if err := rn.finish(t, 10*time.Second); err != nil {
|
|
t.Errorf("the teller stopped with %v, want a clean stop", err)
|
|
}
|
|
}
|
|
|
|
// The principals of the composed accounts file for the teller's tests, with the permissions mesh-controller composes for
|
|
// them (internal/broker): a module's own state is the read grants of its bucket and `$KV.<bucket>.>` to write it
|
|
// (state.go stateGrants, readGrants), a read of another module's bucket is the read grants alone, and a module that uses
|
|
// the seat operator-channel submits its own `accept.ask.<module>`, `accept.cancel.<module>` and `accept.condition.<module>`,
|
|
// hears its own `event.decided.<module>` and reads `$KV.asks.<module>.>` of the router's record (seattraffic.go
|
|
// SeatTrafficOf, with the messenger's seat declaring accepts ask, cancel, condition, emits decided, by-caller and records
|
|
// asks). The gate is a bundle on the tool runner's account (ADR 0437 §3), so the gate's writes are the account node-tools'.
|
|
// The permissions are these lists written out, not the controller's function: the broker tests of issue 1243 prove that.
|
|
func readOf(bucket string) []string {
|
|
stream := "KV_" + bucket
|
|
return []string{"$JS.API.STREAM.INFO." + stream, "$JS.API.DIRECT.GET." + stream + ".>", "$JS.API.CONSUMER.CREATE." + stream + ".>",
|
|
"$JS.API.CONSUMER.DELETE." + stream + ".>", "$JS.FC." + stream + ".>"}
|
|
}
|
|
|
|
func keepsOf(bucket string) []string { return append(readOf(bucket), "$KV."+bucket+".>") }
|
|
|
|
func user(name string, pub, sub []string) string {
|
|
q := func(l []string) string {
|
|
var b strings.Builder
|
|
for _, s := range l {
|
|
b.WriteString(fmt.Sprintf(" %q\n", s))
|
|
}
|
|
return b.String()
|
|
}
|
|
return fmt.Sprintf(` { user: %q, password: "pw"
|
|
permissions {
|
|
publish { allow: [
|
|
%s ] }
|
|
subscribe { allow: [
|
|
%s ] }
|
|
allow_responses: true
|
|
}
|
|
}
|
|
`, name, q(pub), q(append([]string{"_INBOX." + name + ".>"}, sub...)))
|
|
}
|
|
|
|
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"
|
|
seatDecided = "mesh.seat.operator-channel.event.decided.mesh-delivery-teller"
|
|
)
|
|
|
|
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.>")
|
|
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
|
|
jetstream { store_dir: %q }
|
|
accounts {
|
|
MESH {
|
|
jetstream: enabled
|
|
users = [
|
|
{ user: "ctl", password: "pw" }
|
|
%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))
|
|
}
|
|
|
|
type fullBed struct {
|
|
srv *server.Server
|
|
ctl *nats.Conn
|
|
js jetstream.JetStream
|
|
ctx context.Context
|
|
}
|
|
|
|
// startFull starts the bus with the principals above and what the controller makes before any module starts: the buckets
|
|
// of the gate's asks, the teller's answers and asked, the router's work queue SEAT_OPERATOR_CHANNEL.
|
|
func startFull(t *testing.T) *fullBed {
|
|
t.Helper()
|
|
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "t")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { os.RemoveAll(dir) })
|
|
conf := filepath.Join(dir, "bus.conf")
|
|
if err := os.WriteFile(conf, []byte(fullConf(dir)), 0o600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
opts, err := server.ProcessConfigFile(conf)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
opts.NoLog, opts.NoSigs = true, true
|
|
srv, err := server.NewServer(opts)
|
|
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)
|
|
ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(ctl.Close)
|
|
js, err := jetstream.New(ctl)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
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, assignerAsks: delivery.AssignerAsksHistory, viewAnswers: delivery.ViewAnswersHistory} {
|
|
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEAT_OPERATOR_CHANNEL", Subjects: []string{"mesh.seat.operator-channel.accept.>"},
|
|
Retention: jetstream.WorkQueuePolicy}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx}
|
|
}
|
|
|
|
func (b *fullBed) as(t *testing.T, name string) jetstream.JetStream {
|
|
t.Helper()
|
|
c, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo(name, "pw"), nats.CustomInboxPrefix("_INBOX."+name))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(c.Close)
|
|
js, err := jetstream.New(c)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return js
|
|
}
|
|
|
|
// put writes a key as a principal, within a bounded time: a write the permissions refuse reaches the writer as a timeout.
|
|
func (b *fullBed) put(t *testing.T, js jetstream.JetStream, bucket, key string) error {
|
|
t.Helper()
|
|
ctx, cancel := context.WithTimeout(b.ctx, 3*time.Second)
|
|
defer cancel()
|
|
kv, err := js.KeyValue(ctx, bucket)
|
|
if err != nil {
|
|
t.Fatalf("%s cannot be bound: %v", bucket, err)
|
|
}
|
|
_, err = kv.Put(ctx, key, []byte(`{}`))
|
|
return err
|
|
}
|
|
|
|
// TestCredentialsWriteOnlyTheirOwnBuckets: the teller's credential is refused writing the gate's asks, and the gate's
|
|
// (the tool runner's) and the view's are refused writing the teller's answers; each writes where it may, so a refusal is the
|
|
// permissions' and not a bus that does not take writes (1343/AC4, ADR 0437 §6's check).
|
|
func TestCredentialsWriteOnlyTheirOwnBuckets(t *testing.T) { // 1343/AC4
|
|
b := startFull(t)
|
|
teller, view, tools := b.as(t, "n1.mesh-delivery-teller"), b.as(t, "n1.mesh-delivery-view"), b.as(t, "n1.node-tools")
|
|
if err := b.put(t, teller, tellerAnswers, "merge.g.0123456789ab"); err != nil {
|
|
t.Errorf("the teller cannot write its own answers: %v", err)
|
|
}
|
|
if err := b.put(t, tools, gateAsksKV, "merge.g.0123456789ab"); err != nil {
|
|
t.Errorf("the gate's account cannot write the gate's asks: %v", err)
|
|
}
|
|
if err := b.put(t, teller, gateAsksKV, "merge.g.ffffffffffff"); err == nil {
|
|
t.Error("the teller's credential wrote the gate's asks")
|
|
}
|
|
if err := b.put(t, tools, tellerAnswers, "merge.g.ffffffffffff"); err == nil {
|
|
t.Error("the gate's account wrote the teller's answers")
|
|
}
|
|
if err := b.put(t, view, tellerAnswers, "merge.g.ffffffffffff"); err == nil {
|
|
t.Error("the view's credential wrote the teller's answers")
|
|
}
|
|
// What was refused is not there.
|
|
for _, c := range []struct{ bucket, key string }{{gateAsksKV, "merge.g.ffffffffffff"}, {tellerAnswers, "merge.g.ffffffffffff"}} {
|
|
kv, _ := b.js.KeyValue(b.ctx, c.bucket)
|
|
if _, err := kv.Get(b.ctx, c.key); err == nil {
|
|
t.Errorf("%s holds %s though its write was refused", c.bucket, c.key)
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestTellerCarriesAnAskAndWritesTheAnswer starts the program on the composed principal against the bus: the gate's account
|
|
// writes an ask, the teller publishes it to the seat's work queue under its own name, the router (the test) says the
|
|
// warrant on the teller's decided subject, and the answer is under the ask's key in the teller's bucket; withdrawing a
|
|
// second ask cancels it (1343/AC1, AC2).
|
|
func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
|
|
now = time.Now().UTC()
|
|
b := startFull(t)
|
|
tools := b.as(t, "n1.node-tools")
|
|
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, ga := gateAsk(t, twoHeads(), twoHeads())
|
|
kv, err := tools.KeyValue(b.ctx, gateAsksKV)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// The program hears warrants before it reads the asks; give it its subscription before the ask is written.
|
|
time.Sleep(500 * time.Millisecond)
|
|
if _, err := kv.Put(b.ctx, key, value); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seat, err := b.js.Stream(b.ctx, "SEAT_OPERATOR_CHANNEL")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var raw *jetstream.RawStreamMsg
|
|
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
|
if raw, err = seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil {
|
|
break
|
|
}
|
|
}
|
|
if raw == nil {
|
|
t.Fatalf("no ask on the seat's queue: %v\n%s", err, logs.String())
|
|
}
|
|
var asked asks.Ask
|
|
if err := json.Unmarshal(raw.Data, &asked); err != nil || asked.Digest() != ga.Ask.Digest() {
|
|
t.Fatalf("the ask on the queue is %+v (%v), want the gate's", asked, err)
|
|
}
|
|
// The router says the warrant.
|
|
w, _ := json.Marshal(warrantFor(ga.Ask, "approve"))
|
|
if err := b.ctl.Publish(seatDecided, w); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
answers, err := b.js.KeyValue(b.ctx, tellerAnswers)
|
|
if 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 != "approve" {
|
|
t.Errorf("answer %+v err %v", a, err)
|
|
}
|
|
// A second ask, withdrawn by the gate, is cancelled at the router.
|
|
other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}}
|
|
key2, value2, ga2 := gateAsk(t, other, other)
|
|
if _, err := kv.Put(b.ctx, key2, value2); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
|
if _, err := seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil && strings.Contains(logs.String(), ga2.Ask.ID) {
|
|
break
|
|
}
|
|
}
|
|
if err := kv.Delete(b.ctx, key2); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
|
if raw, err = seat.GetLastMsgForSubject(b.ctx, seatCancel); err == nil {
|
|
break
|
|
}
|
|
}
|
|
if err != nil || string(raw.Data) != `{"id":"`+ga2.Ask.ID+`"}` {
|
|
t.Errorf("the withdrawn ask was not cancelled: %v %v\n%s", raw, err, logs.String())
|
|
}
|
|
}
|
|
|
|
// TestTellerAsksAgainAnAskRefusedForWantOfRoom starts the program against the bus: the router (the test) refuses the gate's
|
|
// ask for want of room, as the live router did on 2026-10-21 (issue 1610), and then says another of the teller's asks
|
|
// lapsed; the refused ask is on the seat's work queue again, which the bus's duplicate window would have dropped under its
|
|
// first message id (1610/AC1 on the real bus).
|
|
func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
|
|
now = time.Now().UTC()
|
|
b := startFull(t)
|
|
tools := b.as(t, "n1.node-tools")
|
|
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) })
|
|
kv, err := tools.KeyValue(b.ctx, gateAsksKV)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seat, err := b.js.Stream(b.ctx, "SEAT_OPERATOR_CHANNEL")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
time.Sleep(500 * time.Millisecond)
|
|
// lastAsk waits for the seat's queue to hold an ask newer than seq under id, and gives its sequence.
|
|
lastAsk := func(id string, after uint64) uint64 {
|
|
t.Helper()
|
|
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
|
raw, err := seat.GetLastMsgForSubject(b.ctx, seatAsk)
|
|
if err != nil || raw.Sequence <= after {
|
|
continue
|
|
}
|
|
var a asks.Ask
|
|
if json.Unmarshal(raw.Data, &a) == nil && a.ID == id {
|
|
return raw.Sequence
|
|
}
|
|
}
|
|
t.Fatalf("the ask %s is not on the seat's queue after sequence %d\n%s", id, after, logs.String())
|
|
return 0
|
|
}
|
|
say := func(w asks.Warrant) {
|
|
t.Helper()
|
|
raw, _ := json.Marshal(w)
|
|
if err := b.ctl.Publish(seatDecided, raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
key, value, held := gateAsk(t, twoHeads(), twoHeads())
|
|
if _, err := kv.Put(b.ctx, key, value); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
first := lastAsk(held.Ask.ID, 0)
|
|
say(asks.Warrant{Ask: held.Ask.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeRefused, Words: roomRefusal, At: now})
|
|
for deadline := time.Now().Add(10 * time.Second); !strings.Contains(logs.String(), "for want of room"); time.Sleep(50 * time.Millisecond) {
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("the refusal was not heard\n%s", logs.String())
|
|
}
|
|
}
|
|
other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}}
|
|
key2, value2, ga2 := gateAsk(t, other, other)
|
|
if _, err := kv.Put(b.ctx, key2, value2); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
second := lastAsk(ga2.Ask.ID, first)
|
|
// The other ask lapses at the router: there is room, and the refused ask is asked again.
|
|
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")
|
|
}
|
|
}
|
|
|
|
// TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer starts the program against the bus: the assigner's hold ask, put in
|
|
// its bucket, is on the operator channel's queue as the teller's ask; the router's warrant for "release" is written under the
|
|
// ask's key, where the assigner reads it; a second ask the assigner takes away (a hold it lifted) is cancelled at the router
|
|
// (1906/H1, H2 on the real bus).
|
|
func TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer(t *testing.T) {
|
|
now = time.Now().UTC()
|
|
b := startFull(t)
|
|
assigner := b.as(t, "n1.mesh-delivery-assigner")
|
|
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)
|
|
}
|
|
time.Sleep(500 * time.Millisecond) // the program hears warrants and watches
|
|
if _, err := askKV.Put(b.ctx, key, value); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seat, err := b.js.Stream(b.ctx, "SEAT_OPERATOR_CHANNEL")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var raw *jetstream.RawStreamMsg
|
|
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
|
if raw, err = seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil {
|
|
break
|
|
}
|
|
}
|
|
if raw == nil {
|
|
t.Fatalf("the assigner's ask is not on the seat's queue: %v\n%s", err, logs.String())
|
|
}
|
|
var asked asks.Ask
|
|
if err := json.Unmarshal(raw.Data, &asked); err != nil || asked.ID != ask.ID || asked.Digest() != ask.Digest() {
|
|
t.Fatalf("the ask on the queue is %+v (%v), want the assigner's", asked, err)
|
|
}
|
|
w, _ := json.Marshal(ackFor(ask, "release"))
|
|
if err := b.ctl.Publish(seatDecided, w); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
answers, err := b.js.KeyValue(b.ctx, tellerAnswers)
|
|
if 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)
|
|
}
|
|
|
|
key2, err := delivery.HoldAskKey("plex", "src2:"+strings.Repeat("3", 64))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ask2 := ask
|
|
ask2.ID = delivery.AskID(key2)
|
|
value2, err := delivery.EncodeAssignerAsk(key2, delivery.AssignerAsk{Ask: ask2})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := askKV.Put(b.ctx, key2, value2); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
|
if strings.Contains(logs.String(), ask2.ID+" is asked") {
|
|
break
|
|
}
|
|
}
|
|
if err := askKV.Delete(b.ctx, key2); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
|
if raw, err = seat.GetLastMsgForSubject(b.ctx, seatCancel); err == nil {
|
|
break
|
|
}
|
|
}
|
|
if err != nil || string(raw.Data) != `{"id":"`+ask2.ID+`"}` {
|
|
t.Errorf("the ask taken away was not cancelled: %v %v\n%s", raw, err, logs.String())
|
|
}
|
|
}
|