mesh-delivery/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.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"
	"git.novox.be/novox/mesh-sdk/go/process"

	"mesh-delivery-teller/internal/ends"
)

// 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(process.Account{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.messenger_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
// messenger_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 {
	// ends is the teller's own state (manifest state entry ends), which the controller makes before it starts: a bed
	// without it held the start ten seconds on a read the bus never answered, and the tests raced that stall.
	teller := append(append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), keepsOf(ends.Bucket)...), readOf(gateAsksKV)...),
		seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_"+messengerAsks+".$KV."+messengerAsks+".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, ends.Bucket: 1, messengerAsks: 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}
}

// listening waits until the program hears the router's warrants (its subscription on the decided subject is on the bus),
// which it does before it watches any bucket: an event, not a pause a loaded machine can outlast.
func (b *fullBed) listening(t *testing.T) {
	t.Helper()
	acc, err := b.srv.LookupAccount("MESH")
	if err != nil {
		t.Fatal(err)
	}
	for deadline := time.Now().Add(20 * time.Second); !acc.SubscriptionInterest(seatDecided); time.Sleep(10 * time.Millisecond) {
		if time.Now().After(deadline) {
			t.Fatal("the program did not subscribe to the router's warrants within 20 s")
		}
	}
}

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
	t.Cleanup(nowIs(time.Now().UTC()))
	b := startFull(t)
	tools := b.as(t, "n1.node-tools")
	broker := filepath.Join(t.TempDir(), "broker")
	cred, _ := json.Marshal(process.Account{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) })

	// Issue 774: the router's record is read on the teller's own credential from the bucket the controller grants it.
	if mkv, err := b.js.KeyValue(b.ctx, messengerAsks); err != nil {
		t.Fatal(err)
	} else if _, err := mkv.Put(b.ctx, "mesh-delivery-teller.r1", []byte(`{"state":"open"}`)); err != nil {
		t.Fatal(err)
	}
	tc, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo("n1.mesh-delivery-teller", "pw"), nats.CustomInboxPrefix("_INBOX.n1.mesh-delivery-teller"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(tc.Close)
	if rec, err := recordVia(b.ctx, tc, "mesh-delivery-teller")("r1"); err != nil || rec == nil {
		t.Fatalf("the router's record read as the teller: %v, %v; want it", rec, err)
	}

	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; it has its subscription before the ask is written.
	b.listening(t)
	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
	t.Cleanup(nowIs(time.Now().UTC()))
	b := startFull(t)
	tools := b.as(t, "n1.node-tools")
	broker := filepath.Join(t.TempDir(), "broker")
	cred, _ := json.Marshal(process.Account{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)
	}
	b.listening(t)
	// 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
	t.Cleanup(nowIs(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(process.Account{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)
	}
	b.listening(t) // the program hears warrants and watches
	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 with the same option changes nothing under the ask (another option, later, is the
	// operator changing their mind: TestTellerTakesAChangedHoldAnswerFromTheView, issue 1906 note 23).
	if _, err := viewKV.Put(b.ctx, key+".view", viewAnswer(t, "release", "")); err != nil {
		t.Fatal(err)
	}
	time.Sleep(time.Second)
	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) {
	t.Cleanup(nowIs(time.Now().UTC()))
	b := startFull(t)
	assigner := b.as(t, "n1.mesh-delivery-assigner")
	broker := filepath.Join(t.TempDir(), "broker")
	cred, _ := json.Marshal(process.Account{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)
	}
	b.listening(t) // 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())
	}
}

// TestTellerTakesAChangedHoldAnswerFromTheView starts the program against the bus: a hold ask answered "keep" earlier, and the
// view's answer "release" written now, replace the answer of that asking with "release" (1906/N23 on the real bus).
func TestTellerTakesAChangedHoldAnswerFromTheView(t *testing.T) {
	t.Cleanup(nowIs(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(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	key, value, ask := holdAsk(t)
	answers, err := b.js.KeyValue(b.ctx, tellerAnswers)
	if err != nil {
		t.Fatal(err)
	}
	kept, _ := delivery.EncodeAnswer(delivery.Answer{Option: "keep", Warrant: ask.ID + "@" + time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano), Words: "the operator chose keep"})
	if _, err := answers.Put(b.ctx, delivery.AnswerKey(key), kept); err != nil {
		t.Fatal(err)
	}
	askKV, err := assigner.KeyValue(b.ctx, assignerAsks)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := askKV.Put(b.ctx, key, value); err != nil {
		t.Fatal(err)
	}
	env := map[string]string{"MESH_BROKER_FILE": broker}
	logs := &syncBuffer{}
	rn := startRunner(b.ctx, func(c context.Context) error { return run(c, func(k string) string { return env[k] }, logs) })
	t.Cleanup(func() { rn.finish(t, 10*time.Second) })
	viewKV, err := view.KeyValue(b.ctx, viewAnswers)
	if err != nil {
		t.Fatal(err)
	}
	b.listening(t) // the program hears warrants and watches
	if _, err := viewKV.Put(b.ctx, key+".view", viewAnswer(t, "release", "it was a flake")); err != nil {
		t.Fatal(err)
	}
	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		if e, err := answers.Get(b.ctx, delivery.AnswerKey(key)); err == nil {
			if a, err := delivery.DecodeAnswer(e.Value()); err == nil && a.Option == "release" && answeredAsking(e.Value()) == ask.ID {
				return
			}
		}
	}
	t.Fatalf("the kept hold's answer was not changed to release\n%s", logs.String())
}