jschoubben
ba1e90fb2d
The controller ended (ADR 0590): nothing consumes what nodes report on CONTROL, so the mesh no longer declares the stream, which would make it again before its removal; MeshConsumers is empty, the controller's consumers and the events it followed go, and the lease holder's reading (LeaseHolder, LeaseHandedAway) with them. Only the consumers' names stay, for the removal of the leftovers and the watchman's read of whether controller still stands. Issue 1126 B3b. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
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, its 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
|
|
}
|