package broker 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 }