package main import ( "context" "errors" "fmt" "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" ) // The streams and durable consumers the mesh's own traffic needs, derived once (novox/hq to-be 45 ยง4, // D6 and D7). // // **What the controller asserts at its start and what the self-check expects to find are one // derivation**, run against the bus to make them and against a recorder to list them. Two lists would // drift, and a self-check comparing the bus with a second opinion of what should be there would find // the drift rather than the fault. // assertBusObjects brings every stream and consumer into being on r, and answers the machines that // can now hear a declaration. func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Raiser) ([]string, error) { nodes, err := inv.Nodes(ctx) if err != nil { return nil, err } names := make([]string, 0, len(nodes)) for _, n := range nodes { names = append(names, n.Name) } if err := broker.Raise(r, names); err != nil { return nil, err } // The work queues of the mesh's own roles (novox/hq ADR 0121). The queue before the holder, // deliberately: work queues until somebody arrives to do it, so assigning a build machine a week // after something started asking for builds flushes the backlog instead of having lost it. // With the seats' holders, so each role's work queue gets the consumer its holder takes // work from. Passed as nil until the first live raise, which left the build machine bound to a // consumer nothing had created (2026-09-28). holders, err := seatHolders(ctx, inv) if err != nil { return nil, err } if err := broker.RaiseSeats(r, inventory.MeshSeats(), holders); err != nil { return nil, err } // And how every module hears what it consumes. Derived from the same records the user list is // composed from, so a module the mesh grants a consumer's subjects has that consumer waiting. // Done on every raise, not only when a credential is issued: every module moved onto this bus // by the rollout was issued on the old one, and came up with nothing to bind to (2026-09-28). consumers, err := moduleConsumers(ctx, inv) if err != nil { return nil, err } // Every one tried, and every failure named: one module's consumer the bus refuses is no reason // the modules after it in the list hear nothing (novox/hq issue 208, where this runs on each send). var failed []error for _, c := range consumers { if err := r.EnsureConsumer(c.Consumer); err != nil { failed = append(failed, fmt.Errorf("how %s on %s hears what it consumes: %w", c.Module, c.Node, err)) } } if len(failed) > 0 { return nil, errors.Join(failed...) } return names, nil } // What raises the condition a send says when the objects it implies could not be asserted, and its kind. const ( sourceBusObjects = "bus-objects" kindBusObjectsUnasserted = "bus-objects-unasserted" ) // assertOnSend asserts, on the bus a send is about to use, every object assertBusObjects derives โ€” // **whenever a declaration is sent, not only when the controller starts** (novox/hq issue 208). // // A module assigned after the controller started was sent its declaration and found no consumer to // bind (`consumer not found`, messenger on 2026-10-06), and a seat holder assigned after it found no // worker: the objects a declaration implies were asserted at start and nowhere else, so they existed // only for what was assigned before the last restart. The same derivation, not a second list of what // a send needs: what start asserts, the self-check expects and a send asserts are one answer. Every // part is idempotent, so asserting the whole of it again is the no-op a restart already relies on. // // **A failure is said and raised, and the send goes on.** The objects are the mesh's, not the // machines' being sent: holding every machine back for one consumer that none of them may use would // turn one fault into all of them, and the declarations are not what is wrong. It is never silent โ€” // said in the send's own output and raised as a condition, which the next send that asserts them // clears โ€” and the start-time raise still refuses to serve without them. func assertOnSend(ctx context.Context, inv *inventory.Inventory, r broker.Raiser, indent string) error { _, err := assertBusObjects(ctx, inv, r) var observed []conditions.Observation if err != nil { fmt.Printf("%sTHE BUS DOES NOT HOLD WHAT THIS SEND IMPLIES: %v\n", indent, err) fmt.Printf("%s a module may find no consumer to bind, or a holder no worker; sent anyway, raised as "+ "condition %s, and asserted again by the next send\n", indent, unassertedObservation(err).Key()) observed = append(observed, unassertedObservation(err)) } // Observed when it failed, cleared when it did not: a send that asserted everything is the // observation that the bus holds what it should. if kerr := withKeeper(ctx, func(k *conditions.Keeper) error { return k.Reconcile(ctx, sourceBusObjects, observed) }); kerr != nil { fmt.Printf("%sand whether the bus holds what this send implies could not be kept as a condition: %v\n", indent, kerr) } return err } // unassertedObservation is a send's failure to assert the bus's objects, as a condition. func unassertedObservation(err error) conditions.Observation { return conditions.Observation{Scope: conditions.ScopeBus, ID: "objects", Token: "unasserted", Kind: kindBusObjectsUnasserted, Severity: conditions.Warning, Source: sourceBusObjects, Summary: "the bus's streams and consumers could not be asserted when a declaration was sent: " + "a module may find no consumer to bind, or a seat's holder no worker", Said: err.Error()} } // moduleConsumers is every module's durable consumer, from the records the user list is composed from. func moduleConsumers(ctx context.Context, inv *inventory.Inventory) ([]broker.ModuleConsumer, error) { records, err := inv.BusRecords(ctx) if err != nil { return nil, err } users, err := broker.Users(records) if err != nil { return nil, err } return broker.ConsumersOf(users), nil } // moduleConsumerCount is how many modules hear what they consume, for the raise's one line. func moduleConsumerCount(ctx context.Context, inv *inventory.Inventory) (int, error) { consumers, err := moduleConsumers(ctx, inv) return len(consumers), err } // expectedBusObjects is every stream and consumer assertBusObjects would make, made nowhere. func expectedBusObjects(ctx context.Context, inv *inventory.Inventory) ([]broker.Stream, []broker.Consumer, error) { var rec recordingRaiser if _, err := assertBusObjects(ctx, inv, &rec); err != nil { return nil, nil, fmt.Errorf("what the bus should hold cannot be worked out: %w", err) } return rec.streams, rec.consumers, nil } // recordingRaiser keeps what it was asked to assert and asserts nothing. type recordingRaiser struct { streams []broker.Stream consumers []broker.Consumer } func (r *recordingRaiser) EnsureStream(s broker.Stream) error { r.streams = append(r.streams, s) return nil } func (r *recordingRaiser) EnsureConsumer(c broker.Consumer) error { r.consumers = append(r.consumers, c) return nil }