Node-engine composer: a module new to the node is held until the seat list was read, unless its seats are bootstrapped; the read hands it (H-7, ADR 0636 §1; issue 1966)
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user