mesh-vault: the player holds a family's events until the vault subscribes, and the rotation test waits for the vault's placement

The rotation test failed twice under parallel runs ("never: the rotation begins"), from two races in
its player. The player refused mesh/subscribe, so the placement events it said never reached the
vault: when the vault's first view was answered before the test wrote the placement, the vault never
knew the bus node. And the test wrote the bus node's have right after the placement, so the vault
could hear the have before it applied the placement, adopt nothing, and never learn the loaded list's
revision. The player now answers mesh/subscribe and holds a family's events until it is subscribed,
as the module's consumer does; the test writes the placement only after the vault's first view and
the have only once the vault's picture holds every placement key (vault_first_views), so both orders
are fixed rather than waited out.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig

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

@@ -69,6 +69,64 @@ type player struct {
 	placeMu       sync.Mutex
 	placeValues   map[string]json.RawMessage
 	placeRevision uint64
+	// answeredOnce closes answered at the vault's first view of the placement.
+	answeredOnce sync.Once
+	answered     chan struct{}
+
+	// The families the vault subscribed to, and the events said on one before it did: held, as the module's consumer on
+	// the bus holds them, and handed in order once it subscribes. A runtime that refused the subscription and dropped the
+	// events left the vault's picture at its first view, so a placement written after that view never reached it, and
+	// whether it did turned on which came first (the rotation test's flake, met twice).
+	subMu      sync.Mutex
+	subscribed map[string]bool
+	held       []heldEvent
+}
+
+// heldEvent is an event said on a family the vault has not subscribed to yet.
+type heldEvent struct {
+	key  string
+	body json.RawMessage
+}
+
+// event says an event of key to the vault as its runtime does, holding it until the vault subscribed to its family.
+func (p *player) event(key string, body json.RawMessage) {
+	p.subMu.Lock()
+	defer p.subMu.Unlock()
+	if !p.subscribed[key] {
+		p.held = append(p.held, heldEvent{key: key, body: body})
+		return
+	}
+	p.deliver(key, body)
+}
+
+// deliver sends one event to the vault; under subMu, so events reach it in the order they were said.
+func (p *player) deliver(key string, body json.RawMessage) {
+	p.mu.Lock()
+	p.next++
+	id := fmt.Sprintf("o-%d", p.next)
+	p.mu.Unlock()
+	p.send(map[string]any{"id": id, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{
+		"key": key, "node": "novox", "body": body}}})
+}
+
+// subscribe marks the vault's subscription to pattern (a family's key, as the vault names it) and hands it the events
+// held for it, in order.
+func (p *player) subscribe(pattern string) {
+	p.subMu.Lock()
+	defer p.subMu.Unlock()
+	if p.subscribed == nil {
+		p.subscribed = map[string]bool{}
+	}
+	p.subscribed[pattern] = true
+	rest := p.held[:0]
+	for _, e := range p.held {
+		if e.key == pattern {
+			p.deliver(e.key, e.body)
+			continue
+		}
+		rest = append(rest, e)
+	}
+	p.held = rest
 }
 
 var placeCreated = time.Date(2026, 11, 4, 9, 0, 0, 0, time.UTC)
@@ -86,6 +144,7 @@ func (p *player) followPlacement() {
 		p.t.Fatal(err)
 	}
 	p.placeValues = map[string]json.RawMessage{}
+	p.answered = make(chan struct{})
 	listed := make(chan struct{})
 	go func() {
 		current := true
@@ -112,12 +171,7 @@ func (p *player) followPlacement() {
 					ev["value"] = json.RawMessage(e.Value())
 				}
 				body, _ := json.Marshal(ev)
-				p.mu.Lock()
-				p.next++
-				id := fmt.Sprintf("o-%d", p.next)
-				p.mu.Unlock()
-				p.send(map[string]any{"id": id, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{
-					"key": "mesh-delivery-assigner.views.placement", "node": "novox", "body": json.RawMessage(body)}}})
+				p.event(PlacementViewsEvent, body)
 			}
 			p.placeMu.Unlock()
 		}
@@ -129,6 +183,57 @@ func (p *player) followPlacement() {
 	}
 }
 
+// firstViewAnswered waits until the vault's first view of the placement is answered, so what the test writes after it
+// reaches the vault only as the owner's events.
+func (p *player) firstViewAnswered() {
+	p.t.Helper()
+	select {
+	case <-p.answered:
+	case <-time.After(30 * time.Second):
+		p.t.Fatal("the vault never asked its first view of the placement")
+	}
+}
+
+// placementHeld waits until the vault's picture of the placement holds every key the placement's bucket holds now:
+// the owner said each change, and the vault applied it (its first-views tool says the revision its picture holds). What
+// the vault does on another state's change (a have) reads the placement from that picture, so a test writes that change
+// only after this, never racing the owner's events.
+func (p *player) placementHeld() {
+	p.t.Helper()
+	kv, err := p.js.KeyValue(p.ctx, "mesh-delivery-assigner_placement")
+	if err != nil {
+		p.t.Fatal(err)
+	}
+	keys, err := kv.Keys(p.ctx)
+	if err != nil {
+		p.t.Fatal(err)
+	}
+	var want uint64
+	eventually(p.t, "the owner said every key of the placement", func() bool {
+		p.placeMu.Lock()
+		defer p.placeMu.Unlock()
+		for _, k := range keys {
+			if _, ok := p.placeValues[k]; !ok {
+				return false
+			}
+		}
+		want = p.placeRevision
+		return true
+	})
+	eventually(p.t, "the vault's picture of the placement holds every key", func() bool {
+		raw, err := p.call(FirstViewsTool, "person.jochen", nil)
+		if err != nil {
+			return false
+		}
+		var a struct {
+			FirstViews []struct {
+				Revision uint64 `json:"revision"`
+			} `json:"first-views"`
+		}
+		return json.Unmarshal(raw, &a) == nil && len(a.FirstViews) == 1 && a.FirstViews[0].Revision >= want
+	})
+}
+
 // 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 {
@@ -282,19 +387,30 @@ func (p *player) handle(m wire) {
 		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))
+	case "mesh/subscribe":
+		var sub struct {
+			Pattern string `json:"pattern"`
+		}
+		_ = json.Unmarshal(m.Params, &sub)
+		p.answer(m.ID, map[string]any{}, nil)
+		p.subscribe(sub.Pattern)
 	case "mesh/ask":
 		var ask struct {
 			Key  string          `json:"key"`
 			Body json.RawMessage `json:"body"`
 		}
 		_ = json.Unmarshal(m.Params, &ask)
-		if ask.Key == PlacementTool && p.placeValues != nil {
+		p.placeMu.Lock()
+		following := p.placeValues != nil
+		p.placeMu.Unlock()
+		if ask.Key == PlacementTool && following {
 			p.placeMu.Lock()
 			answer := map[string]any{"v": 1, "owner": "mesh-delivery-assigner", "state": "placement", "revision": p.placeRevision,
 				"created": placeCreated, "values": p.placeValues, "at": time.Now()}
 			raw, _ := json.Marshal(answer)
 			p.placeMu.Unlock()
 			p.answer(m.ID, json.RawMessage(raw), nil)
+			p.answeredOnce.Do(func() { close(p.answered) })
 			return
 		}
 		if p.ask != nil {

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

@@ -93,7 +93,9 @@ func TestTheVaultWholeRotatesANodesLoginWithAStandInEngine(t *testing.T) {
 
 	// The bus node (novox, running nats) and the node that rotates (g14); no lease key anywhere (issue 1126: the controller
 	// ended by its unassignment); g14's login adopted from
-	// the loaded list, its password known to the engine.
+	// the loaded list, its password known to the engine. Written after the vault's first view of the placement, so they
+	// reach it only as the owner's views events, whichever the vault would otherwise have read first.
+	p.firstViewAnswered()
 	for node, modules := range map[string][]string{"novox": {"nats"}, "g14": {"web"}} {
 		pl := delivery.Placement{V: delivery.Version, Node: node}
 		for _, m := range modules {
@@ -110,6 +112,7 @@ func TestTheVaultWholeRotatesANodesLoginWithAStandInEngine(t *testing.T) {
 			}
 		}
 	}
+	p.placementHeld()
 	oldPassword := "the-old-password-of-g14"
 	oldHash, _ := bcrypt.GenerateFromPassword([]byte(oldPassword), bcrypt.MinCost)
 	p.put(t, "mesh-vault.bus-accounts", "node.g14", AccountHash{User: "node.g14", Hash: string(oldHash), Adopted: true})