diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 6607fc6..0bc5d21 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -102,6 +102,18 @@ type Env struct { ComposedState ComposedStore // Now is the clock what the composer waits on is timed by; nil is time.Now. Now func() time.Time + // SeatList is the node's reader of the seat list (novox/hq ADR 0636 §1): until the list was read, the composer hands + // nothing new of a module whose seats are not all bootstrapped, and keeps what runs of it. Nil gates nothing. + SeatList SeatGate + // SeatListChanged says the reader's held set changed: the node composes again, so what waited for the list is + // handed at the read and not on a timer. Nil hears nothing. + SeatListChanged <-chan struct{} +} + +// SeatGate is what the composer asks of the reader of the seat list: whether it may compose a module that claims, +// invokes or provides these seats (seatlist.Reader.MayAct). +type SeatGate interface { + MayAct(seats ...string) bool } // VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value. @@ -439,6 +451,25 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er } }() } + if c.env.SeatListChanged != nil { + go func() { + for { + select { + case <-ctx.Done(): + return + case _, ok := <-c.env.SeatListChanged: + if !ok { + return + } + select { + case merged <- tagged{in: -1, ok: true}: + case <-ctx.Done(): + return + } + } + } + }() + } out := make(chan agent.Update, 64) c.watching.Store(true) go func() { @@ -908,6 +939,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { done() return nil } + // A reader that has not read the seat list acts on nothing new (ADR 0636 §1): a module whose composition differs from + // what runs, and whose seats are not all bootstrapped, is held with what it runs. + held = c.holdUntilListed(ctx, r, compositions, held, running) // The boxes it runs, slot by slot: read before c.mu is taken, since the agent's lock is taken to read what runs. missed = c.missed.Load() for key, comp := range compositions { @@ -1731,3 +1765,78 @@ func mapValues(m map[string]delivery.Composed) []delivery.Composed { } return out } + +// WaitsForTheSeatList is why a module is held while the node has not read the seat list. +const WaitsForTheSeatList = "the node has not read the seat list, so it composes nothing new of this module and keeps what runs of it (novox/hq ADR 0636 §1)" + +// holdUntilListed takes out of compositions every module's composition that is new (no running composition has its +// digest) while the reader may not act on its seats, and names the module in held: it keeps running what it runs, and +// the pass after the list was read hands it. The node's own composition and a module whose composition runs already are +// left as they are. +func (c *Composer) holdUntilListed(ctx context.Context, r reader, compositions map[string]delivery.Composed, held map[string]string, running []delivery.Composed) map[string]string { + if c.env.SeatList == nil { + return held + } + ran := map[string]bool{} + for _, comp := range running { + if dig, err := delivery.ComposedDigestOf(comp); err == nil { + ran[comp.Module+"/"+dig] = true + } + } + for key, comp := range compositions { + if comp.Module == "" { + continue + } + if dig, err := delivery.ComposedDigestOf(comp); err == nil && ran[comp.Module+"/"+dig] { + continue + } + if c.env.SeatList.MayAct(c.seatsOf(ctx, r, comp.Module)...) { + continue + } + delete(compositions, key) + if held == nil { + held = map[string]string{} + } + if _, already := held[comp.Module]; !already { + held[comp.Module] = WaitsForTheSeatList + } + } + return held +} + +// seatsOf is the seats a module claims, invokes or provides, from the manifest its module record names; none when it +// cannot be read, which the reader judges as a module no bootstrapped seat names. +func (c *Composer) seatsOf(ctx context.Context, r reader, module string) []string { + c.mu.Lock() + m, ok := c.modules[module] + c.mu.Unlock() + if !ok || m.Manifest == "" { + return nil + } + raw, err := r.Manifest(ctx, m.Manifest) + if err != nil { + return nil + } + var man struct { + Claims []struct { + Name string `json:"name"` + } `json:"claims"` + Invokes []string `json:"invokes"` + Provides []struct { + Name string `json:"name"` + } `json:"provides"` + } + if json.Unmarshal(raw, &man) != nil { + return nil + } + var seats []string + for _, cl := range man.Claims { + seats = append(seats, cl.Name) + } + for _, in := range man.Invokes { + if seat, _, ok := strings.Cut(strings.TrimPrefix(in, "seat:"), "."); ok && strings.HasPrefix(in, "seat:") { + seats = append(seats, seat) + } + } + return seats +} diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 170a0b3..542667a 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -1750,3 +1750,84 @@ func setFor[T any](t *testing.T, p *T, v T) { *p = v t.Cleanup(func() { *p = old }) } + +type gate struct{ open atomic.Bool } + +func (g *gate) MayAct(...string) bool { return g.open.Load() } + +// ADR 0636 §1: a node that has not read the seat list composes nothing new of a module whose seats are not bootstrapped, +// says it holds the module (so its have says why) and hands it at the read, which the reader says as an event; what +// already runs with the same composition is not held. +func TestAnUnreadSeatListHoldsWhatIsNewAndHandsItAtTheRead(t *testing.T) { + b := newFakeBus() + fp := delivery.SourcePrefix2 + strings.Repeat("a", 64) + digest := delivery.ManifestDigest([]byte(lemurs)) + man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: lemurs}) + key, _ := delivery.ManifestsKey(digest) + b.kv[delivery.ManifestsBucket+"/"+key] = man + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + env := b.env() + g := &gate{} + changed := make(chan struct{}, 1) + env.SeatList, env.SeatListChanged = g, changed + var mu sync.Mutex + heldBy := map[string]string{} + env.Refused = func(_ string, held map[string]string) { mu.Lock(); heldBy = held; mu.Unlock() } + c := New(env) + out, err := c.Watch(ctx, "n1.>") + if err != nil { + t.Fatal(err) + } + place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}, + Modules: []delivery.PlacedModule{{Module: "lemurs"}}}) + have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}) + mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}) + want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true}) + nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true}) + b.put(t, delivery.PlacementBucket, "placement.n1", place) + b.put(t, delivery.HaveBucket, "n1.node", have) + b.put(t, delivery.ModulesBucket, "lemurs", mod) + b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want) + b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant) + for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} { + b.send(bucket, agent.Update{Caught: true}) + } + handed := map[string]bool{} + for { + u := next(t, out) + if u.Caught { + break + } + handed[u.Key] = true + } + mu.Lock() + why := heldBy["lemurs"] + mu.Unlock() + if why != WaitsForTheSeatList { + t.Errorf("lemurs is held with %q, want %q", why, WaitsForTheSeatList) + } + for k := range handed { + if strings.Contains(k, "lemurs") { + t.Errorf("the node handed %s before it read the seat list", k) + } + } + if !handed["n1.node"] { + t.Errorf("the node's own composition was held with the module: %v", handed) + } + // The reader reads the list and says so: the module is handed, with no timer. + g.open.Store(true) + changed <- struct{}{} + for { + u := next(t, out) + if strings.Contains(u.Key, "lemurs") && !u.Deleted { + break + } + } + mu.Lock() + still := heldBy["lemurs"] + mu.Unlock() + if still != "" { + t.Errorf("lemurs is still held after the read: %q", still) + } +}