diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 5311d3a..54791b2 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -305,6 +305,9 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } } takePlacement := func(a appeared) error { + if kvs[a.name] != nil { + return nil // bound already, by the other of its two events + } kvs[a.name] = a.kv logger.Printf("the bucket %s is open and watched", a.name) return watch(a.name) @@ -343,6 +346,23 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } case <-dirty: reopenPuts(p) + // The grants part makes the seat's buckets in the pass that seals the list granting them, so their creation can + // be said before this credential may hear it: every pass, itself started by a watched change, binds each one the + // bus now holds, and says the seat waits while one is missing. + var unbound []string + for _, name := range seatBuckets { + if kvs[name] != nil || js == nil { + continue + } + if kv, err := js.KeyValue(ctx, name); err == nil { + if err := takePlacement(appeared{name, kv}); err != nil { + return err + } + continue + } + unbound = append(unbound, name) + } + p.sayWaits(unbound) p.pass(ctx) } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 651b73a..a1f264f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -48,6 +48,8 @@ type passer struct { seat *seatActs publish publisher saidProposed map[string]bool + // saidWaits is the last said of the seat's wait for its states (sayWaits). + saidWaits string intended map[string]map[string]map[string]bool } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go index 79b00f3..85d363d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go @@ -15,6 +15,7 @@ import ( "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/catalogue" "git.novox.be/novox/mesh-sdk/go/compose" "git.novox.be/novox/mesh-sdk/go/compose/frombus" @@ -180,6 +181,23 @@ func (p *passer) withCalls(pairs []assign.Pair) []assign.Pair { return pairs } +// sayWaits raises, once, that the seat waits for the states not yet bound, and clears it once, when all are. +func (p *passer) sayWaits(unbound []string) { + c := assign.SeatWaitsCondition(unbound) + state := string(c.State) + c.Summary + if p.say == nil || p.said[c.Key] && p.saidWaits == state { + return + } + if c.State == asks.StateCleared && p.saidWaits == "" { + return // nothing was raised by this process + } + if err := p.say(c); err != nil { + p.logf("that the seat waits for its states could not be said, and is said at the next pass: %v", err) + return + } + p.said[c.Key], p.saidWaits = true, state +} + // seatPlacements is every node's placement key and its revision, as the watch holds them. func (p *passer) seatPlacements() (map[string]delivery.Placement, map[string]uint64) { out, revs := map[string]delivery.Placement{}, map[string]uint64{} diff --git a/modules/mesh-delivery-assigner/internal/assign/placement.go b/modules/mesh-delivery-assigner/internal/assign/placement.go index d7eaa3c..e164ef9 100644 --- a/modules/mesh-delivery-assigner/internal/assign/placement.go +++ b/modules/mesh-delivery-assigner/internal/assign/placement.go @@ -2,6 +2,7 @@ package assign import ( "fmt" + "strings" "sort" "git.novox.be/novox/mesh-sdk/go/asks" @@ -104,3 +105,17 @@ func PlacementCondition(p Pair, raised bool) asks.Condition { "name a module on a node (ADR 0549 §5): the assigner writes no want for it there and leaves the one it wrote before, so "+ "nothing moves or is removed. This clears once both name it or neither does.", asks.NeedsYou, needs, p.Only, p.Module, p.Node)} } + +// SeatWaitsCondition is the assigner's word on the seat mesh-placement while a state of the seat is not bound (ADR 0567 +// §1): raised naming each, cleared when none is left. +func SeatWaitsCondition(unbound []string) asks.Condition { + key := delivery.AssignerModule + ".seat-states" + subject := asks.Scope{Scope: "module", ID: delivery.AssignerModule} + if len(unbound) == 0 { + return asks.Condition{Key: key, Subject: subject, State: asks.StateCleared, Resolved: "the seat mesh-placement holds all its states."} + } + why := "the seat waits for its states: " + strings.Join(unbound, ", ") + " are not on the bus or not granted yet" + return asks.Condition{Key: key, Severity: asks.SeverityWarning, Subject: subject, State: asks.StateRaised, + Summary: "the seat waits for its states", Headline: "the seat waits for its states", + Explanation: asks.NothingToDo + " " + why + "; each is bound at the next pass once it is, with no restart."} +} diff --git a/modules/mesh-delivery-assigner/internal/assign/seat_test.go b/modules/mesh-delivery-assigner/internal/assign/seat_test.go index e478e8c..2a94a8a 100644 --- a/modules/mesh-delivery-assigner/internal/assign/seat_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/seat_test.go @@ -82,6 +82,15 @@ func TestThePlacementSeatDecides(t *testing.T) { t.Error("a proposal written whole leaves something to write") } + // The seat says it waits while one of its states is not bound, naming it, and clears once all are. + for _, unbound := range [][]string{{delivery.PlacementPulsesBucket}, nil} { + c := SeatWaitsCondition(unbound) + if _, err := asks.EncodeCondition(c); err != nil || (len(unbound) > 0) != (c.State == asks.StateRaised) || + (len(unbound) > 0 && !strings.Contains(c.Explanation, "the seat waits for its states: "+delivery.PlacementPulsesBucket)) { + t.Errorf("waiting for %v: %+v, %v", unbound, c, err) + } + } + writes, err := Writes(p, laid) if err != nil || len(writes) != 2 || writes[0].Key != "placement.anchor" || writes[0].Base != 4 || writes[1].Base != 9 { t.Fatalf("writes = %+v, %v", writes, err)