control-plane: publish mesh events for real

cmd/sam-control-plane/main.go

@@ -46,6 +46,7 @@ var (
 	biscuitTTL            time.Duration
 	oidcSessionTTL        time.Duration
 	nodeRetention         time.Duration
+	meshReconnectInterval time.Duration
 	adminTokenPath        string
 	insecureSkipTLSVerify bool
 	logLevel              string
@@ -142,6 +143,19 @@ func main() {
 				}
 			}()
 
+			// Bans, key rotations and policy updates reach the mesh as they
+			// happen; every consumer also pulls, so this is speed, not truth.
+			mesh, err := controlplane.NewMeshPublisher(cmd.Context(), store, meshReconnectInterval)
+			if err != nil {
+				logger.Fatalf("Failed to start mesh event publisher: %v", err)
+			}
+			defer func() {
+				if err := mesh.Close(); err != nil {
+					logger.Errorf("Failed to stop mesh event publisher: %v", err)
+				}
+			}()
+			srv.SetMeshAdapter(mesh)
+
 			if err := srv.Start(); err != nil {
 				logger.Fatalf("Failed to start control plane: %v", err)
 			}
@@ -164,6 +178,7 @@ func main() {
 	rootCmd.Flags().DurationVar(&biscuitTTL, "biscuit-ttl", api.BiscuitTokenTTL, "Lifespan minted into every issued Biscuit's expiration fact. Capped to the OIDC token's own expiry when shorter.")
 	rootCmd.Flags().DurationVar(&oidcSessionTTL, "oidc-session-ttl", api.OIDCSessionTTL, "How long an OIDC enrollment stays refreshable before the identity must re-authenticate with the OIDC provider. Shorter values keep the provider authoritative for offboarding at the cost of more frequent interactive re-enrollment.")
 	rootCmd.Flags().DurationVar(&nodeRetention, "node-retention", controlplane.DefaultNodeRetention, "How long an enrolled node's record is kept after its session expired before it is deleted. Banned nodes are always kept. 0 keeps every record forever.")
+	rootCmd.Flags().DurationVar(&meshReconnectInterval, "mesh-reconnect-interval", controlplane.DefaultMeshReconnectInterval, "How often the event publisher re-reads the router leases and dials any router it is not connected to.")
 	rootCmd.Flags().StringVar(&adminTokenPath, "admin-token-path", "", "Path to file containing the token for authenticating policy REST API requests (or env SAM_ADMIN_TOKEN)")
 	rootCmd.Flags().BoolVar(&insecureSkipTLSVerify, "insecure-skip-tls-verify", false, "Skip TLS verification for OIDC providers")
 	rootCmd.Flags().StringVar(&logLevel, "log-level", "info", "Log level (debug, info, warn, error)")

internal/controlplane/mesh.go

@@ -22,10 +22,12 @@ import (
 	"sync"
 	"time"
 
+	"github.com/libp2p/go-libp2p"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
-	"github.com/libp2p/go-libp2p/core/peerstore"
+	"github.com/multiformats/go-multiaddr"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/google/sam/api"
@@ -87,42 +89,161 @@ func (n *NopMeshAdapter) Close() error {
 	return nil
 }
 
-// P2PMeshAdapter implements MeshAdapter using a libp2p Host and GossipSub subscriber/publisher.
+// P2PMeshAdapter publishes control plane events on the mesh's gossip topic.
+//
+// It is the control plane's whole presence on the mesh, and it is one-way:
+// events go out so a ban, a key rotation or a policy change reaches routers
+// and nodes the moment it happens, and nothing is read back. Every consumer
+// still pulls /keys, /info and /policies on its own schedule, so a missed
+// event is a delay, never a divergence. Where the adapter runs on a host of
+// its own (NewMeshPublisher) that host has no listen address, no DHT, no
+// relay and no stream handlers: it can dial routers and nothing can dial it.
 type P2PMeshAdapter struct {
 	host  host.Host
-	ps    *pubsub.PubSub
 	topic *pubsub.Topic
 	store storage.Store
 	mu    sync.Mutex
+	// close tears down what NewMeshPublisher built; nil when the host and
+	// topic belong to someone else (sam-one's embedded router).
+	close func() error
+}
+
+// NewP2PMeshAdapter publishes on an existing host's topic. The caller owns
+// both and closes them; Close on the adapter is a no-op.
+func NewP2PMeshAdapter(h host.Host, topic *pubsub.Topic, store storage.Store) (*P2PMeshAdapter, error) {
+	if h == nil || topic == nil || store == nil {
+		return nil, fmt.Errorf("host, topic, and store cannot be nil")
+	}
+	if topic.String() != api.GossipEvents {
+		return nil, fmt.Errorf("topic %q is not the mesh events topic %q", topic.String(), api.GossipEvents)
+	}
+	return &P2PMeshAdapter{host: h, topic: topic, store: store}, nil
 }
 
-func NewP2PMeshAdapter(h host.Host, ps *pubsub.PubSub, store storage.Store) (*P2PMeshAdapter, error) {
-	if h == nil || ps == nil || store == nil {
-		return nil, fmt.Errorf("host, pubsub, and store cannot be nil")
+// RouterDialTimeout bounds each attempt to reach a leased router.
+const RouterDialTimeout = 10 * time.Second
+
+// DefaultMeshReconnectInterval is how often the publisher re-reads the lease
+// table; a router that just enrolled waits at most this long for events. It
+// is one query and at most a few dials per tick, so it is kept short.
+const DefaultMeshReconnectInterval = 30 * time.Second
+
+// NewMeshPublisher builds the control plane's own publish-only peer and
+// keeps it connected to every router holding a lease, re-checking the lease
+// table every reconnect. The lease table is all the control plane needs to
+// know about the mesh's shape, and it already has it. Close stops the loop
+// and the host.
+func NewMeshPublisher(ctx context.Context, store storage.Store, reconnect time.Duration) (*P2PMeshAdapter, error) {
+	if store == nil {
+		return nil, fmt.Errorf("store cannot be nil")
+	}
+	if reconnect <= 0 {
+		return nil, fmt.Errorf("reconnect interval must be positive, got %s", reconnect)
+	}
+	h, err := libp2p.New(libp2p.NoListenAddrs, libp2p.DisableRelay())
+	if err != nil {
+		return nil, fmt.Errorf("mesh publisher host: %w", err)
+	}
+	loopCtx, cancel := context.WithCancel(ctx)
+	// StrictSign is the default; pinned because routers and nodes key their
+	// per-author rate limit on the signed sender.
+	ps, err := pubsub.NewGossipSub(loopCtx, h, pubsub.WithMessageSignaturePolicy(pubsub.StrictSign))
+	if err != nil {
+		cancel()
+		_ = h.Close()
+		return nil, fmt.Errorf("mesh publisher gossipsub: %w", err)
 	}
 	topic, err := ps.Join(api.GossipEvents)
 	if err != nil {
-		return nil, fmt.Errorf("failed to join gossip events topic %s: %w", api.GossipEvents, err)
+		cancel()
+		_ = h.Close()
+		return nil, fmt.Errorf("join %s: %w", api.GossipEvents, err)
+	}
+	p := &P2PMeshAdapter{host: h, topic: topic, store: store}
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		p.keepRoutersConnected(loopCtx, reconnect)
+	}()
+	p.close = func() error {
+		cancel()
+		wg.Wait()
+		_ = topic.Close()
+		return h.Close()
 	}
+	logger.Infof("[Mesh] Publishing control plane events as %s", h.ID())
+	return p, nil
+}
 
-	return &P2PMeshAdapter{
-		host:  h,
-		ps:    ps,
-		topic: topic,
-		store: store,
-	}, nil
+// keepRoutersConnected dials, now and every interval, each leased router the
+// host is not connected to. A publish only reaches peers that have announced
+// the topic, so connections are kept warm ahead of the events rather than
+// made when one is due.
+func (p *P2PMeshAdapter) keepRoutersConnected(ctx context.Context, interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		p.connectRouters(ctx)
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
+func (p *P2PMeshAdapter) connectRouters(ctx context.Context) {
+	routers, err := p.store.GetActiveRouters(ctx)
+	if err != nil {
+		logger.Warnf("[Mesh] Cannot list routers to publish to: %v", err)
+		return
+	}
+	for _, r := range routers {
+		info, err := routerAddrInfo(r)
+		if err != nil {
+			logger.Warnf("[Mesh] Skipping router lease %q: %v", r.PeerID, err)
+			continue
+		}
+		if p.host.Network().Connectedness(info.ID) == network.Connected {
+			continue
+		}
+		dialCtx, cancel := context.WithTimeout(ctx, RouterDialTimeout)
+		err = p.host.Connect(dialCtx, info)
+		cancel()
+		if err != nil {
+			logger.Warnf("[Mesh] Router %s unreachable for event publishing: %v", info.ID, err)
+			continue
+		}
+		logger.Infof("[Mesh] Connected to router %s", info.ID)
+	}
 }
 
-func (p *P2PMeshAdapter) ConnectPeer(ctx context.Context, targetAddr string) error {
-	info, err := peer.AddrInfoFromString(targetAddr)
+// routerAddrInfo is the dial target for a lease. Routers announce their
+// addresses with a trailing /p2p/<id>; the dialer wants the id once and the
+// addresses bare, and an address naming a different peer is not this
+// router's.
+func routerAddrInfo(r storage.RouterLease) (peer.AddrInfo, error) {
+	pid, err := peer.Decode(r.PeerID)
 	if err != nil {
-		return fmt.Errorf("invalid peer address string %q: %w", targetAddr, err)
+		return peer.AddrInfo{}, fmt.Errorf("invalid peer ID: %w", err)
 	}
-	p.host.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.PermanentAddrTTL)
-	if err := p.host.Connect(ctx, *info); err != nil {
-		return fmt.Errorf("failed to connect to peer %s: %w", info.ID, err)
+	info := peer.AddrInfo{ID: pid}
+	for _, s := range r.Addresses {
+		ma, err := multiaddr.NewMultiaddr(s)
+		if err != nil {
+			continue
+		}
+		addr, id := peer.SplitAddr(ma)
+		if addr == nil || (id != "" && id != pid) {
+			continue
+		}
+		info.Addrs = append(info.Addrs, addr)
 	}
-	return nil
+	if len(info.Addrs) == 0 {
+		return peer.AddrInfo{}, fmt.Errorf("no dialable address in %v", r.Addresses)
+	}
+	return info, nil
 }
 
 func (p *P2PMeshAdapter) PublishEvent(ctx context.Context, eventType api.MeshEvent_Type, peerID string, payload []byte) error {
@@ -218,11 +339,8 @@ func (p *P2PMeshAdapter) GetNodeStatus(ctx context.Context, peerID string) (*Nod
 func (p *P2PMeshAdapter) Close() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	if p.topic != nil {
-		_ = p.topic.Close()
+	if p.close == nil {
+		return nil
 	}
-	if p.host != nil {
-		return p.host.Close()
-	}
-	return nil
+	return p.close()
 }