jschoubben
b5f8604d03
The grants part composes the bus's user list with the composer the controller uses (ADR 0547 §5), copied rather than moved: the controller is not changed and keeps its own until it loses the lease. Left out: the bus client, the broker's address from the environment and the client's two methods; the lease's two subjects are copied in. One table test holds the copy to four meshes' lists and the server golden, written by the controller's own internal/broker from the same fixtures (testdata/controller-user-lists.json).
74 lines
2.6 KiB
Go
74 lines
2.6 KiB
Go
package buscompose
|
|
|
|
import "fmt"
|
|
|
|
// Bringing the bus's own objects into being, in the one order that works.
|
|
//
|
|
// **Asserted on every start rather than created once at genesis.** A stream somebody deleted, a mesh
|
|
// raised from a restored backup, or a bus whose data directory was replaced all have records and no
|
|
// objects — and a node whose consumer is missing hears nothing while everything else about it looks
|
|
// correct. Idempotence is the whole requirement, and the parts are already idempotent; this is the
|
|
// order they have to be asked in.
|
|
|
|
// Raiser is everything asserting the bus's objects needs of a connection to it.
|
|
type Raiser interface {
|
|
Asserter
|
|
Ensurer
|
|
}
|
|
|
|
// Raise asserts the mesh's streams, the controller's own consumers, and one consumer per node.
|
|
//
|
|
// **The order is not a preference.** A consumer on a stream that does not exist is refused, and the
|
|
// refusal names the stream rather than the order — so somebody reading it goes looking for a deleted
|
|
// stream instead of a reversed pair of lines. Nodes last, because the one a node reads lives on a
|
|
// stream the mesh's own set defines.
|
|
func Raise(r Raiser, nodes []string) error {
|
|
if err := AssertMeshStreams(r); err != nil {
|
|
return err
|
|
}
|
|
if err := AssertMeshConsumers(r); err != nil {
|
|
return err
|
|
}
|
|
if err := AssertNodeConsumers(r, nodes); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// RaiseSeats asserts one work queue per declared seat, and the worker of whoever holds it.
|
|
//
|
|
// Separate from Raise because it is answered by a different question: the mesh's own objects exist
|
|
// because the mesh does, and a seat's exist because a module declaring one was registered. Kept
|
|
// beside it so the order is visible — a holder's worker needs the seat's stream, and a seat's stream
|
|
// needs nothing.
|
|
func RaiseSeats(r Raiser, seats []DeclaredSeat, holders map[string]Holder) error {
|
|
for _, s := range SeatStreams(seats) {
|
|
if err := r.EnsureStream(s); err != nil {
|
|
return fmt.Errorf("asserting the work queue for %s: %w", s.Name, err)
|
|
}
|
|
}
|
|
for _, s := range seats {
|
|
h, held := holders[s.Name]
|
|
if !held {
|
|
// **The stream exists and the consumer does not, on purpose.** Work queues until a
|
|
// holder appears, so installing the module a week after something started sending to it
|
|
// flushes the backlog instead of having lost it.
|
|
continue
|
|
}
|
|
c, needed := HolderConsumerFor(h.Node, h.Module, s)
|
|
if !needed {
|
|
continue
|
|
}
|
|
if err := r.EnsureConsumer(c); err != nil {
|
|
return fmt.Errorf("asserting how %s on %s works %s: %w", h.Module, h.Node, s.Name, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Holder is which module on which machine holds a seat.
|
|
type Holder struct {
|
|
Node string
|
|
Module string
|
|
}
|