Compare commits
14
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3907ea0db0 | ||
|
|
40f5e9a41c | ||
|
|
2c2eb51878 | ||
|
|
aa2d0b51ea | ||
|
|
64d154d9d7 | ||
|
|
ffa390f916 | ||
|
|
4d62e6caf1 | ||
|
|
386ae676ca | ||
|
|
f8a9c3d6bc | ||
|
|
83671fae5f | ||
|
|
9b715524a2 | ||
|
|
e06fc1ed16 | ||
|
|
13d7c5c5dd | ||
|
|
84024cbdb6 |
@@ -37,7 +37,7 @@ func askCommand(ctx context.Context, args []string) error {
|
||||
arguments = json.RawMessage(positionals[2])
|
||||
}
|
||||
|
||||
server, err := link.Connect(nil, nil)
|
||||
server, err := connectLink(ctx, nil, nil, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -368,7 +368,7 @@ func buildOne(ctx context.Context, source buildSource, path, ref string, wait ti
|
||||
}
|
||||
defer ident.Close()
|
||||
|
||||
server, err := link.Connect(nil, nil)
|
||||
server, err := connectLink(ctx, nil, nil, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -472,7 +472,7 @@ func buildAndShow(ctx context.Context, source buildSource, path, ref string, wai
|
||||
return err
|
||||
}
|
||||
defer ident.Close()
|
||||
server, err := link.Connect(nil, nil)
|
||||
server, err := connectLink(ctx, nil, nil, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -537,6 +537,15 @@ func onTheNetwork(ctx context.Context, inv *inventory.Inventory,
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// **With the seat holders on record**, or a machine running the next holder of a seat beside
|
||||
// the current one resolves as two holders, is refused, and drops out of the map — taking the
|
||||
// address every other machine composes for what it offers (novox/hq ADR 0131). Found live:
|
||||
// the control node vanished from the private network the moment the new bus was assigned
|
||||
// beside the old one.
|
||||
holdings, err := inv.Holdings(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []inventory.Overlay
|
||||
for _, p := range places {
|
||||
if p.Address == "" {
|
||||
@@ -549,7 +558,7 @@ func onTheNetwork(ctx context.Context, inv *inventory.Inventory,
|
||||
caps, _ := inv.ProfileOf(ctx, p.Name)
|
||||
got, err := catalogue.Resolve(shelf, assigned,
|
||||
catalogue.Node{Name: p.Name, Site: p.Site, Capabilities: caps},
|
||||
catalogue.World{Unchecked: true})
|
||||
catalogue.World{Unchecked: true, Holdings: holdings})
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
+35
-13
@@ -38,6 +38,33 @@ func reportUnhostable(node string, plan catalogue.Resolution) {
|
||||
// nothing in it was wrong, and no one edit was the one that should have been a new file.
|
||||
|
||||
// serve is the control plane running: one connection to the broker, one queue, one consumer.
|
||||
// connectLink opens the controller's link over whichever bus this process is on (design 25: one
|
||||
// variable moves it). The streams and this controller's consumers are raised first on the new bus,
|
||||
// so nothing served here finds them missing.
|
||||
func connectLink(ctx context.Context, inv *inventory.Inventory, enroller link.Enroller, listener link.Listener) (*link.Server, error) {
|
||||
busAddress, onNATS, err := broker.OnNATS()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := broker.MustBeOneBus(os.Getenv(broker.AMQPVarName), busAddress); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !onNATS {
|
||||
return link.Connect(enroller, listener)
|
||||
}
|
||||
if inv != nil {
|
||||
if err := raiseTheBus(ctx, inv, busAddress); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
js, err := broker.Dial(busAddress)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the mesh is on the bus at %s and this control plane cannot reach it: %w",
|
||||
busAddress, err)
|
||||
}
|
||||
return link.ConnectNats(js, enroller, listener), nil
|
||||
}
|
||||
|
||||
func serve(ctx context.Context) error {
|
||||
open, err := openStores(ctx)
|
||||
if err != nil {
|
||||
@@ -90,7 +117,7 @@ func serve(ctx context.Context) error {
|
||||
|
||||
work := link.Enrolment{Inventory: inv, Identity: ident, Management: management, Broker: known,
|
||||
OnNATS: onNATS}
|
||||
server, err := link.Connect(work, work)
|
||||
server, err := connectLink(ctx, inv, work, work)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -100,11 +127,6 @@ func serve(ctx context.Context) error {
|
||||
// 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.
|
||||
if onNATS {
|
||||
if err := raiseTheBus(ctx, inv, busAddress); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// And build results nobody was waiting for. A build triggered any other way than `build`
|
||||
// would otherwise be reported into the void, which is the same as not reporting it.
|
||||
server.Records(builds{inv})
|
||||
@@ -158,13 +180,13 @@ func declare(ctx context.Context, args []string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
server, err := link.Connect(nil, nil)
|
||||
server, err := connectLink(ctx, nil, nil, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer server.Close()
|
||||
|
||||
if err := link.Declare(ctx, link.OverCurrent{Channel: server.Channel()}, ident, node, raw, 15*time.Second); err != nil {
|
||||
if err := link.Declare(ctx, server.Bus(), ident, node, raw, 15*time.Second); err != nil {
|
||||
return err
|
||||
}
|
||||
fmt.Printf("sent %s a signed declaration (%d bytes)\n", node, len(raw))
|
||||
@@ -271,7 +293,7 @@ func pushCommand(ctx context.Context, args []string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
server, err := link.Connect(nil, nil)
|
||||
server, err := connectLink(ctx, nil, nil, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -332,7 +354,7 @@ func pushCommand(ctx context.Context, args []string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := link.Declare(ctx, link.OverCurrent{Channel: server.Channel()}, ident, s.node, body, 15*time.Second); err != nil {
|
||||
if err := link.Declare(ctx, server.Bus(), ident, s.node, body, 15*time.Second); err != nil {
|
||||
return err
|
||||
}
|
||||
// After it is away, not before. A digest recorded for something that failed to send would
|
||||
@@ -415,7 +437,7 @@ func pushCommand(ctx context.Context, args []string) error {
|
||||
return declarationWith(held, open, node, plan, settings, gens, Allocating)
|
||||
},
|
||||
func(s readyNode, body []byte) error {
|
||||
if err := link.Declare(ctx, link.OverCurrent{Channel: server.Channel()}, ident, s.node, body,
|
||||
if err := link.Declare(ctx, server.Bus(), ident, s.node, body,
|
||||
15*time.Second); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -628,7 +650,7 @@ func sendTo(ctx context.Context, open *stores, names []string) error {
|
||||
len(refusals), strings.Join(refusals, "\n\n"))
|
||||
}
|
||||
|
||||
server, err := link.Connect(nil, nil)
|
||||
server, err := connectLink(ctx, nil, nil, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -639,7 +661,7 @@ func sendTo(ctx context.Context, open *stores, names []string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := link.Declare(ctx, link.OverCurrent{Channel: server.Channel()}, ident, s.node, body, 15*time.Second); err != nil {
|
||||
if err := link.Declare(ctx, server.Bus(), ident, s.node, body, 15*time.Second); err != nil {
|
||||
return err
|
||||
}
|
||||
record, err := inv.NodeByName(ctx, s.node)
|
||||
|
||||
@@ -340,6 +340,14 @@ func rolloutMint(ctx context.Context, again bool) error {
|
||||
machines++
|
||||
|
||||
case broker.KindModule:
|
||||
if p.Module == "mesh-controller" {
|
||||
// The control plane is a module too, and its `broker` secret is the old bus's
|
||||
// credential it is still using while this runs. Writing the new bus's blob there
|
||||
// cut the mesh off from its own old bus mid-move (2026-09-28). Its new-bus credential
|
||||
// is the controller principal's `bus` secret above; nothing else is needed here.
|
||||
skipped++
|
||||
continue
|
||||
}
|
||||
m, inShelf := shelf[p.Module]
|
||||
if !inShelf {
|
||||
skipped++
|
||||
|
||||
@@ -244,7 +244,14 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
case KindNode:
|
||||
// A host publishes its own node's control traffic and subscribes its own declaration —
|
||||
// and nothing of any other node's.
|
||||
pub = []string{"mesh.control." + p.Node + ".>"}
|
||||
// And binding to its consumer, which asks the server about it (CONSUMER.INFO) — the one
|
||||
// thing the host does that nothing granted. Found the first time a machine dialled a
|
||||
// permissioned server: "this node cannot read its declarations" (2026-09-28). The ack and
|
||||
// the inbox are granted below with every principal's.
|
||||
pub = []string{
|
||||
"mesh.control." + p.Node + ".>",
|
||||
"$JS.API.CONSUMER.INFO.NODES." + p.Node,
|
||||
}
|
||||
sub = []string{"mesh.node." + p.Node + ".declare"}
|
||||
|
||||
case KindModule:
|
||||
|
||||
+1
-1
@@ -32,7 +32,7 @@ accounts {
|
||||
subscribe: { allow: ["_INBOX.enrol.one.>"] }
|
||||
} }
|
||||
{ user: "node.one", password: "$2a$11$nnnnnnnnnnnnnnnnnnnnnn", permissions: {
|
||||
publish: { allow: ["$JS.ACK.NODES.one.>", "mesh.control.one.>"] }
|
||||
publish: { allow: ["$JS.ACK.NODES.one.>", "$JS.API.CONSUMER.INFO.NODES.one", "mesh.control.one.>"] }
|
||||
subscribe: { allow: ["_INBOX.node.one.>", "mesh.node.one.declare"] }
|
||||
} }
|
||||
{ user: "one.telegram", password: "$2a$11$tttttttttttttttttttttt", permissions: {
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
)
|
||||
|
||||
// OverNats is the controller's outbound on the bus being built: the same three acts the other
|
||||
// transport has, on the subjects the permissions were derived for (design 25). A declaration is a
|
||||
// JetStream publish into NODES, where the node's own consumer waits for it; an event is announced on
|
||||
// the subject its name derives to; a tool is asked by request and reply on the module's tool subject.
|
||||
type OverNats struct{ JS *broker.JetStream }
|
||||
|
||||
// declareSubject is where one node's declaration lands — the NODES stream's subject for it, and the
|
||||
// only subject that node's consumer delivers. The host subscribes exactly this.
|
||||
func declareSubject(node string) string { return "mesh.node." + node + ".declare" }
|
||||
|
||||
func (b OverNats) PublishDeclaration(ctx context.Context, node string, body []byte) error {
|
||||
publish, cancel := context.WithTimeout(ctx, 15*time.Second)
|
||||
defer cancel()
|
||||
if _, err := b.JS.Context().Publish(declareSubject(node), body, nats.Context(publish)); err != nil {
|
||||
return fmt.Errorf("declaring to %s: %w", node, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b OverNats) PublishEvent(ctx context.Context, key, source, node string, body []byte) error {
|
||||
// The key is the subject: the controller's own events are named in full, and what a module
|
||||
// emits is derived before it reaches here. Headers carry the envelope the other transport put
|
||||
// in message properties (ADR 0042), so a consumer reads who and when without the payload.
|
||||
msg := nats.NewMsg(key)
|
||||
msg.Data = body
|
||||
msg.Header.Set("x-source", source)
|
||||
msg.Header.Set("x-node", node)
|
||||
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
|
||||
if err := b.JS.Conn().PublishMsg(msg); err != nil {
|
||||
return fmt.Errorf("announcing %s: %w", key, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b OverNats) AskTool(ctx context.Context, module, tool string, args []byte, timeout time.Duration) ([]byte, error) {
|
||||
ask, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
reply, err := b.JS.Conn().RequestWithContext(ask, "mesh.mod."+module+".tool."+tool, args)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("asking %s.%s: %w", module, tool, err)
|
||||
}
|
||||
return reply.Data, nil
|
||||
}
|
||||
|
||||
// ConnectNats is Connect for the bus being built: the controller's inbound and outbound over one
|
||||
// JetStream connection the caller has already raised the streams on. Nothing is declared here —
|
||||
// the streams and the controller's consumers are asserted by Raise, before anything is served.
|
||||
func ConnectNats(js *broker.JetStream, enroller Enroller, listener Listener) *Server {
|
||||
return &Server{
|
||||
inbound: Nats(js),
|
||||
bus: OverNats{JS: js},
|
||||
js: js,
|
||||
enroller: enroller,
|
||||
listener: listener,
|
||||
log: newLog(),
|
||||
}
|
||||
}
|
||||
|
||||
// Bus is the controller's outbound, whichever transport it connected over. Callers that send a
|
||||
// declaration or ask a tool use this rather than the channel, which one transport does not have.
|
||||
func (s *Server) Bus() Bus { return s.bus }
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"log"
|
||||
"os"
|
||||
"time"
|
||||
@@ -70,6 +71,7 @@ type Server struct {
|
||||
bus Bus
|
||||
conn *amqp.Connection
|
||||
channel *amqp.Channel
|
||||
js *broker.JetStream
|
||||
|
||||
enroller Enroller
|
||||
listener Listener
|
||||
@@ -180,10 +182,12 @@ func Connect(enroller Enroller, listener Listener) (*Server, error) {
|
||||
channel: channel,
|
||||
enroller: enroller,
|
||||
listener: listener,
|
||||
log: log.New(os.Stdout, "", log.LstdFlags),
|
||||
log: newLog(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func newLog() *log.Logger { return log.New(os.Stdout, "", log.LstdFlags) }
|
||||
|
||||
// Channel is the controller's channel, for the command line's own publishing.
|
||||
func (s *Server) Channel() *amqp.Channel { return s.channel }
|
||||
|
||||
@@ -197,6 +201,9 @@ func (s *Server) Close() {
|
||||
if s.conn != nil {
|
||||
_ = s.conn.Close()
|
||||
}
|
||||
if s.js != nil {
|
||||
s.js.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// Serve acts on what arrives until the context ends.
|
||||
|
||||
+3
-3
@@ -47,15 +47,14 @@
|
||||
"MESH_STORE_INVENTORY_FILE": "/run/secrets/inventory",
|
||||
"MESH_STORE_IDENTITY_FILE": "/run/secrets/identity",
|
||||
"MESH_STORE_LICENCES_FILE": "/run/secrets/licences",
|
||||
"MESH_BROKER_AMQP_FILE": "/run/secrets/broker",
|
||||
"MESH_BROKER_MANAGEMENT_FILE": "/run/secrets/broker-management",
|
||||
"MESH_BROKER_ADDRESS_FILE": "/run/secrets/broker-address",
|
||||
"MESH_STORE_INVENTORY_PORT": "${seat:mesh-store:5432}",
|
||||
"MESH_STORE_IDENTITY_PORT": "${seat:mesh-store:5432}",
|
||||
"MESH_STORE_LICENCES_PORT": "${seat:mesh-store:5432}",
|
||||
"MESH_BROKER_AMQP_PORT": "${seat:mesh-broker:5672}",
|
||||
"MESH_BROKER_MANAGEMENT_PORT": "${seat:mesh-broker:15672}",
|
||||
"MESH_BROKER_ADDRESS_PORT": "${seat:mesh-broker:5671}"
|
||||
"MESH_BROKER_ADDRESS_PORT": "${seat:mesh-broker:5671}",
|
||||
"MESH_BUS_NATS_FILE": "/run/secrets/bus"
|
||||
},
|
||||
"volumes": [
|
||||
"/var/lib/mesh-broker-tls:/broker-tls:ro",
|
||||
@@ -63,6 +62,7 @@
|
||||
"/var/lib/mesh/mesh-controller/identity:/run/secrets/identity:ro",
|
||||
"/var/lib/mesh/mesh-controller/licences:/run/secrets/licences:ro",
|
||||
"/var/lib/mesh/mesh-controller/broker:/run/secrets/broker:ro",
|
||||
"/var/lib/mesh/mesh-controller/bus:/run/secrets/bus:ro",
|
||||
"/var/lib/mesh/mesh-controller/broker-management:/run/secrets/broker-management:ro",
|
||||
"/var/lib/mesh/mesh-controller/broker-address:/run/secrets/broker-address:ro"
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user