mesh-vault: prove the vault whole on a real bus mints only from the lease write, and try a failed mint again at the pass; adopt the operator key before the watches open (issue 774, ADR 0561 §1-§4)

modules/mesh-vault/cmd/mesh-vault/main.go

@@ -217,6 +217,8 @@ func main() {
 			written, failed := env.Vault.HashHeld()
 			say("hashed %d bus accounts at start; %d could not be, tried again every minute: %v", len(written), len(failed), failed)
 		}
+		// The operator key first: nothing is kept without it, so a want heard before it would not be minted.
+		env.pass()
 		watched := watchHandings(env)
 		lease := watchLease(env)
 		wants := watchWants(env)
@@ -233,6 +235,7 @@ func main() {
 			env.pass()
 			if env.Vault != nil {
 				env.Vault.HashOwed()
+				env.Vault.MintOwed()
 				env.Vault.RetirePass()
 			}
 			env.pairPass()

modules/mesh-vault/cmd/mesh-vault/mint.go

@@ -54,6 +54,8 @@ type minting struct {
 	wants map[string]buscompose.WantedAccount
 	// heldBack are the wanted users the vault did not mint, with why (vault_status).
 	heldBack map[string]string
+	// owed are the wanted users whose mint failed, tried again at the pass (MintOwed).
+	owed map[string]bool
 	// minted are the accounts this vault minted since it started, by user (vault_status).
 	minted map[string]Minted
 	// loadedRevision is the highest revision of the user list the bus node said it loaded (rotation.go).
@@ -168,6 +170,14 @@ func (e errHeldBack) Error() string { return e.why }
 // mintWanted mints w when minting is on, the vault holds no value for it and adopted no hash of it; otherwise it says
 // why not and, where that is a hold, keeps it for vault_status.
 func (v *Vault) mintWanted(w buscompose.WantedAccount, by string) (Minted, error) {
+	m, err := v.mintOnce(w, by)
+	var held errHeldBack
+	v.owe(w.User, err != nil && !errors.As(err, &held) && w.Valid() == nil)
+	return m, err
+}
+
+// mintOnce is one try of mintWanted.
+func (v *Vault) mintOnce(w buscompose.WantedAccount, by string) (Minted, error) {
 	v.minting.mintMu.Lock()
 	defer v.minting.mintMu.Unlock()
 	if err := w.Valid(); err != nil {
@@ -255,6 +265,42 @@ func (v *Vault) mintWanted(w buscompose.WantedAccount, by string) (Minted, error
 	return out, nil
 }
 
+// MintOwed tries again, at the vault's minute pass, every wanted account whose mint failed (its value could not be kept,
+// its key read, its password made): a want heard is never lost to one refusal. A want held back is not owed: the switch
+// or the want changing judges it again.
+func (v *Vault) MintOwed() {
+	v.minting.mu.Lock()
+	var owed []string
+	for u := range v.minting.owed {
+		owed = append(owed, u)
+	}
+	v.minting.mu.Unlock()
+	sort.Strings(owed)
+	for _, u := range owed {
+		if w, ok := v.wanted(u); ok {
+			if _, err := v.mintWanted(w, "the pass, after a mint that failed"); err != nil {
+				v.log("%s is still not minted: %v", u, err)
+			}
+		} else {
+			v.owe(u, false)
+		}
+	}
+}
+
+// owe marks user's mint as failed (tried again at the pass) or not.
+func (v *Vault) owe(user string, owed bool) {
+	v.minting.mu.Lock()
+	defer v.minting.mu.Unlock()
+	if !owed {
+		delete(v.minting.owed, user)
+		return
+	}
+	if v.minting.owed == nil {
+		v.minting.owed = map[string]bool{}
+	}
+	v.minting.owed[user] = true
+}
+
 // heldBackOrMinted is a mint's answer as a tool gives it.
 func heldBackOrMinted(m Minted, err error) (any, error) {
 	var held errHeldBack

modules/mesh-vault/cmd/mesh-vault/mint_integration_test.go

@@ -0,0 +1,422 @@
+//go:build integration
+
+package main
+
+// The vault whole against a real bus (novox/hq ADR 0561 §1 to §4, issue 774's acceptance 12): the built binary, played
+// over its real stdin and stdout as the node's tool runtime plays it (tools/call with the caller in _meta; mesh/state.*
+// and mesh/publish answered from a real nats-server's key-value buckets and subjects), its store in a directory of its
+// own. A want written to the grants part's state is minted only once the controller's lease is handed away: the hash is
+// in bus-accounts, own answers the value sealed to the node, the value is the controller's shape and its hash's
+// password, account-issued is said; an adopted user is never minted, and a lease handed back mints nothing.
+
+import (
+	"bufio"
+	"context"
+	"crypto/rand"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"golang.org/x/crypto/bcrypt"
+	"golang.org/x/crypto/nacl/box"
+
+	"git.novox.be/novox/mesh-sdk/go/buscompose"
+	"git.novox.be/novox/mesh-sdk/go/seal"
+	"git.novox.be/novox/mesh-sdk/go/testbus"
+)
+
+// wire is one line between the runtime and the bundle (the SDK's stdio message).
+type wire struct {
+	ID     json.RawMessage `json:"id,omitempty"`
+	Method string          `json:"method,omitempty"`
+	Params json.RawMessage `json:"params,omitempty"`
+	Result json.RawMessage `json:"result,omitempty"`
+	Error  *struct {
+		Code    int    `json:"code"`
+		Message string `json:"message"`
+	} `json:"error,omitempty"`
+}
+
+// player is the node's tool runtime as far as the vault needs it, over a real bus.
+type player struct {
+	t   *testing.T
+	ctx context.Context
+	nc  *nats.Conn
+	js  jetstream.JetStream
+
+	mu      sync.Mutex
+	in      io.Writer
+	next    int
+	pending map[string]chan wire
+	said    []string
+}
+
+// bucketOf is the bucket a state names: the vault's own by its local name, another's as `<owner>.<name>`.
+func bucketOf(state string) string {
+	if owner, name, ok := strings.Cut(state, "."); ok {
+		return owner + "_" + name
+	}
+	return "mesh-vault_" + state
+}
+
+func (p *player) send(m map[string]any) {
+	raw, _ := json.Marshal(m)
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	_, _ = p.in.Write(append(raw, '\n'))
+}
+
+func (p *player) answer(id json.RawMessage, result any, err error) {
+	if err != nil {
+		p.send(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
+		return
+	}
+	p.send(map[string]any{"id": id, "result": result})
+}
+
+func (p *player) kv(state string) (jetstream.KeyValue, error) {
+	return p.js.KeyValue(p.ctx, bucketOf(state))
+}
+
+// serve answers what the bundle asks, and hands the answers to what the test asked it.
+func (p *player) serve(out io.Reader) {
+	scanner := bufio.NewScanner(out)
+	scanner.Buffer(make([]byte, 1<<20), 1<<24)
+	for scanner.Scan() {
+		var m wire
+		if json.Unmarshal(scanner.Bytes(), &m) != nil {
+			continue
+		}
+		if m.Method == "" {
+			var key string
+			_ = json.Unmarshal(m.ID, &key)
+			p.mu.Lock()
+			ch := p.pending[key]
+			delete(p.pending, key)
+			p.mu.Unlock()
+			if ch != nil {
+				ch <- m
+			}
+			continue
+		}
+		go p.handle(m)
+	}
+}
+
+func (p *player) handle(m wire) {
+	var a struct {
+		State string          `json:"state"`
+		Key   string          `json:"key"`
+		Value json.RawMessage `json:"value"`
+		Body  json.RawMessage `json:"body"`
+	}
+	_ = json.Unmarshal(m.Params, &a)
+	switch m.Method {
+	case "mesh/state.get":
+		kv, err := p.kv(a.State)
+		if err != nil {
+			p.answer(m.ID, nil, err)
+			return
+		}
+		e, err := kv.Get(p.ctx, a.Key)
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			p.answer(m.ID, nil, nil)
+			return
+		}
+		if err != nil {
+			p.answer(m.ID, nil, err)
+			return
+		}
+		p.answer(m.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil)
+	case "mesh/state.put":
+		kv, err := p.kv(a.State)
+		if err == nil {
+			var rev uint64
+			rev, err = kv.Put(p.ctx, a.Key, a.Value)
+			if err == nil {
+				p.answer(m.ID, map[string]any{"revision": rev}, nil)
+				return
+			}
+		}
+		p.answer(m.ID, nil, err)
+	case "mesh/state.keys":
+		kv, err := p.kv(a.State)
+		if err != nil {
+			p.answer(m.ID, nil, err)
+			return
+		}
+		keys, err := kv.Keys(p.ctx)
+		if errors.Is(err, jetstream.ErrNoKeysFound) {
+			keys, err = []string{}, nil
+		}
+		p.answer(m.ID, keys, err)
+	case "mesh/state.delete":
+		kv, err := p.kv(a.State)
+		if err == nil {
+			err = kv.Delete(p.ctx, a.Key)
+		}
+		p.answer(m.ID, map[string]any{}, err)
+	case "mesh/state.watch":
+		kv, err := p.kv(a.State)
+		if err != nil {
+			p.answer(m.ID, nil, err)
+			return
+		}
+		key := strings.ReplaceAll(a.Key, "**", ">")
+		if key == "" {
+			key = ">"
+		}
+		w, err := kv.Watch(p.ctx, key)
+		if err != nil {
+			p.answer(m.ID, nil, err)
+			return
+		}
+		var watch string
+		_ = json.Unmarshal(m.ID, &watch)
+		current := true
+		for e := range w.Updates() {
+			if e == nil {
+				current = false
+				p.answer(m.ID, map[string]any{}, nil)
+				continue
+			}
+			op := "put"
+			if e.Operation() != jetstream.KeyValuePut {
+				op = "delete"
+			}
+			change := map[string]any{"state": a.State, "key": e.Key(), "op": op, "revision": e.Revision(), "current": current}
+			if op == "put" {
+				change["value"] = json.RawMessage(e.Value())
+			}
+			p.mu.Lock()
+			p.next++
+			id := fmt.Sprintf("w-%d", p.next)
+			p.mu.Unlock()
+			p.send(map[string]any{"id": id, "method": "mesh/state", "params": map[string]any{"watch": watch, "change": change}})
+		}
+	case "mesh/publish":
+		var e struct {
+			Key  string          `json:"key"`
+			Body json.RawMessage `json:"body"`
+		}
+		_ = json.Unmarshal(m.Params, &e)
+		p.mu.Lock()
+		p.said = append(p.said, e.Key+" "+string(e.Body))
+		p.mu.Unlock()
+		p.answer(m.ID, map[string]any{}, p.nc.Publish("mesh.mod.mesh-vault.event."+e.Key, e.Body))
+	default:
+		// mesh/ask and the rest: nothing answers on this bus, as a call with no responder.
+		p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
+	}
+}
+
+// call is a tools/call of the vault as caller, answered as the tool's JSON.
+func (p *player) call(name, caller string, args map[string]any) (json.RawMessage, error) {
+	p.mu.Lock()
+	p.next++
+	id := fmt.Sprintf("c-%d", p.next)
+	ch := make(chan wire, 1)
+	p.pending[id] = ch
+	p.mu.Unlock()
+	p.send(map[string]any{"id": id, "method": "tools/call", "params": map[string]any{"name": name, "arguments": args,
+		"_meta": map[string]any{"mesh/caller": caller}}})
+	select {
+	case m := <-ch:
+		var r struct {
+			Content []struct {
+				Text string `json:"text"`
+			} `json:"content"`
+			IsError bool `json:"isError"`
+		}
+		if m.Error != nil {
+			return nil, errors.New(m.Error.Message)
+		}
+		if err := json.Unmarshal(m.Result, &r); err != nil || len(r.Content) != 1 {
+			return nil, fmt.Errorf("%s answered %s", name, m.Result)
+		}
+		if r.IsError {
+			return nil, errors.New(r.Content[0].Text)
+		}
+		return json.RawMessage(r.Content[0].Text), nil
+	case <-time.After(30 * time.Second):
+		return nil, fmt.Errorf("%s did not answer in 30s", name)
+	}
+}
+
+func (p *player) put(t *testing.T, state, key string, v any) {
+	t.Helper()
+	kv, err := p.kv(state)
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw, _ := json.Marshal(v)
+	if _, err := kv.Put(p.ctx, key, raw); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (p *player) hashOf(t *testing.T, user string) *AccountHash {
+	t.Helper()
+	kv, err := p.kv(BusAccountsState)
+	if err != nil {
+		t.Fatal(err)
+	}
+	e, err := kv.Get(p.ctx, user)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return nil
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+	var a AccountHash
+	if err := json.Unmarshal(e.Value(), &a); err != nil {
+		t.Fatal(err)
+	}
+	return &a
+}
+
+func eventually(t *testing.T, what string, ok func() bool) {
+	t.Helper()
+	for end := time.Now().Add(30 * time.Second); time.Now().Before(end); time.Sleep(50 * time.Millisecond) {
+		if ok() {
+			return
+		}
+	}
+	t.Fatalf("never: %s", what)
+}
+
+// 774/AC12.
+func TestTheVaultWholeMintsOnTheBusOnlyFromTheLeaseWrite(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	dir := t.TempDir()
+	bin := filepath.Join(dir, "mesh-vault")
+	build := exec.Command("go", "build", "-o", bin, ".")
+	build.Stderr = os.Stderr
+	if err := build.Run(); err != nil {
+		t.Fatalf("the vault does not build: %v", err)
+	}
+	nc, err := nats.Connect(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, b := range []string{"mesh-vault_bus-accounts", "mesh-grants_accounts", "mesh-controller_lease",
+		"mesh-controller_assignments", "node-engine_have", "mesh-delivery-indexer_placement"} {
+		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 2}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// The operator key, as the mesh exports it to the vault's root.
+	opPub, _, err := box.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	root := filepath.Join(dir, "root")
+	_ = os.MkdirAll(root, 0o700)
+	export, _ := json.Marshal(map[string]string{"operator-key": base64.StdEncoding.EncodeToString(opPub[:])})
+	if err := os.WriteFile(filepath.Join(root, "export.json"), export, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	cmd := exec.CommandContext(ctx, bin)
+	cmd.Env = append(os.Environ(), "MESH_VAULT_STORE="+filepath.Join(dir, "store"), "MESH_VAULT_STATE="+filepath.Join(dir, "state"),
+		"MESH_VAULT_ROOT="+root, "MESH_VAULT_LEDGER="+filepath.Join(dir, "ledger"), "MESH_RECEIVES="+filepath.Join(dir, "mesh.json"),
+		"MESH_NODE=novox", "MESH_MODULE=mesh-vault")
+	stdin, _ := cmd.StdinPipe()
+	stdout, _ := cmd.StdoutPipe()
+	cmd.Stderr = os.Stderr
+	if err := cmd.Start(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() })
+	p := &player{t: t, ctx: ctx, nc: nc, js: js, in: stdin, pending: map[string]chan wire{}}
+	go p.serve(stdout)
+
+	status := func() MintState {
+		raw, err := p.call("vault_status", "person.jochen", nil)
+		if err != nil {
+			return MintState{}
+		}
+		var s struct {
+			Minting MintState `json:"minting"`
+		}
+		_ = json.Unmarshal(raw, &s)
+		return s.Minting
+	}
+	// The controller holds the lease: a want is held back, nothing is minted.
+	p.put(t, "mesh-controller.lease", LeaseKey, map[string]any{"instance": "novox/1/2026-10-27", "epoch": 57})
+	x, y, z := aWant("mod-x"), aWant("mod-y"), aWant("mod-z")
+	p.put(t, "mesh-vault.bus-accounts", y.User, AccountHash{User: y.User, Hash: "$2a$04$adoptedadoptedadoptedadoptedadoptedadoptedadopte", Adopted: true})
+	p.put(t, "mesh-grants.accounts", x.User, x)
+	p.put(t, "mesh-grants.accounts", y.User, y)
+	eventually(t, "the want held back under the controller's lease", func() bool {
+		return strings.Contains(status().HeldBack[x.User], "controller holds the lease")
+	})
+	if p.hashOf(t, x.User) != nil {
+		t.Fatal("minted under the controller's lease")
+	}
+	// The write that hands the lease away mints it.
+	p.put(t, "mesh-controller.lease", LeaseKey, map[string]any{"handed-to": "the operator's hand-over"})
+	eventually(t, "the account minted at the lease write", func() bool { return p.hashOf(t, x.User) != nil })
+	nodePub, nodePriv := aSealingKey(t)
+	if _, err := p.call(Seat+".node-key", "node.ace", map[string]any{"key": nodePub}); err != nil {
+		t.Fatal(err)
+	}
+	raw, err := p.call(Seat+".own", "node.ace", map[string]any{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var own OwnAnswer
+	if err := json.Unmarshal(raw, &own); err != nil {
+		t.Fatal(err)
+	}
+	if len(own.Own) != 1 || own.Own[0].Module != "mod-x" || own.Own[0].Name != "broker" || own.Own[0].Origin != OriginMade {
+		t.Fatalf("own answers %+v", own.Own)
+	}
+	value, err := seal.Open(own.Own[0].Sealed, nodePriv)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var held struct{ Password string }
+	_ = json.Unmarshal([]byte(value), &held)
+	shape, _ := buscompose.HeldCredential(x, held.Password)
+	if value != string(shape) {
+		t.Errorf("the minted value %q is not the controller's shape %q", value, shape)
+	}
+	if h := p.hashOf(t, x.User); bcrypt.CompareHashAndPassword([]byte(h.Hash), []byte(held.Password)) != nil || h.Adopted || h.Generation != 1 {
+		t.Errorf("bus-accounts %s holds %+v, not the minted password's hash", x.User, h)
+	}
+	if h := p.hashOf(t, y.User); h == nil || !h.Adopted {
+		t.Errorf("the adopted %s was replaced: %+v", y.User, h)
+	}
+	p.mu.Lock()
+	said := strings.Join(p.said, "\n")
+	p.mu.Unlock()
+	if !strings.Contains(said, AccountIssuedEvent+` {"`) || !strings.Contains(said, `"user":"ace.mod-x"`) || strings.Contains(said, held.Password) {
+		t.Errorf("the vault said: %s", said)
+	}
+	// Handed back to a controller: a new want mints nothing.
+	p.put(t, "mesh-controller.lease", LeaseKey, map[string]any{"instance": "novox/2/2026-10-28", "epoch": 58})
+	p.put(t, "mesh-grants.accounts", z.User, z)
+	eventually(t, "the new want held back once the lease is the controller's again", func() bool {
+		return strings.Contains(status().HeldBack[z.User], "controller holds the lease")
+	})
+	if p.hashOf(t, z.User) != nil {
+		t.Error("minted after the lease was handed back")
+	}
+}

modules/mesh-vault/cmd/mesh-vault/mint_test.go

@@ -221,3 +221,24 @@ func TestOneRotationBeginsPerRevisionOfTheLoadedList(t *testing.T) {
 		t.Error("a covered account rotated in place")
 	}
 }
+
+// A mint that failed (here: no operator key yet, so nothing can be kept) is tried again at the pass, never lost.
+func TestAFailedMintIsTriedAgainAtThePass(t *testing.T) {
+	v, accounts, _ := mintingVault(t, true)
+	key, _ := v.Store.OperatorKey()
+	if err := v.Store.setOperatorKey("", "the test", v.Now()); err != nil {
+		t.Fatal(err)
+	}
+	w := aWant("mod-x")
+	v.WantHeard(w.User, wantRaw(t, w), false)
+	if _, ok := accounts.keys[w.User]; ok {
+		t.Fatal("minted with no operator key")
+	}
+	if err := v.Store.setOperatorKey(key, "the test", v.Now()); err != nil {
+		t.Fatal(err)
+	}
+	v.MintOwed()
+	if _, ok := accounts.keys[w.User]; !ok {
+		t.Errorf("not minted at the pass after the failure: %+v", v.MintState())
+	}
+}