diff --git a/modules/mesh-vault/cmd/mesh-vault/mint_integration_test.go b/modules/mesh-vault/cmd/mesh-vault/mint_integration_test.go index 1ec932400..a8295d826 100644 --- a/modules/mesh-vault/cmd/mesh-vault/mint_integration_test.go +++ b/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 `.`. 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 { diff --git a/modules/mesh-vault/cmd/mesh-vault/rotation_integration_test.go b/modules/mesh-vault/cmd/mesh-vault/rotation_integration_test.go index fcb44b18b..b1bfaaf2f 100644 --- a/modules/mesh-vault/cmd/mesh-vault/rotation_integration_test.go +++ b/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})