jschoubben
4de10e32e3
Two of 1.7's three remaining pieces. **Raised on every start, not 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. The order is not a preference: a consumer on a stream that does not exist is refused *naming the stream*, so somebody reading that refusal goes looking for a deletion instead of a reversed pair of lines. Pinned by a test, along with the one thing about seats that reads like an omission and is not — a seat's work queue is asserted whether or not anybody holds it, because work queues until a holder appears, so installing the module a week later flushes the backlog instead of having lost it. Against a real server: every object accepted, asserting twice changes nothing (a start that failed the second time is a controller that cannot restart), a machine joining an already-raised bus is accepted, each node's consumer is bound to its own declaration subject and no other's, and CONTROL does not dead-letter — because the store window's bound belongs to the controller and a server that gave up first would discard the push the stream exists to protect. **Which bus this mesh is on is one fact, read in one place.** Every seam the change went behind ships both implementations; this is what the rollout flips. Being told about both is refused at start rather than warned about: a mesh half on each is one where a declaration goes out on one bus and the report comes back on the other, and every component logs success while it happens — ADR 0074's failure arriving through configuration instead of through code. The refusal names both variables and says which to unset, because whoever reads it has to choose and the wrong choice is a rollout half done.
74 lines
2.6 KiB
Go
74 lines
2.6 KiB
Go
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
|
|
}
|