control-plane: publish mesh events for real - #454
Conversation
There was a problem hiding this comment.
Code Review
This pull request implements a publish-only mesh publisher for the control plane, allowing it to periodically discover and dial leased routers to propagate events like bans, key rotations, and policy updates. It introduces the --mesh-reconnect-interval flag, refactors P2PMeshAdapter to manage these active connections, and adds corresponding unit tests. The review feedback suggests two key improvements: dialing unconnected routers concurrently to prevent slow or offline routers from blocking the reconnect loop, and making the Close method of P2PMeshAdapter idempotent to avoid potential issues from multiple invocations.
| 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) | ||
| } |
There was a problem hiding this comment.
Dialing routers sequentially can block the reconnect loop for a significant amount of time if multiple routers are offline or slow to respond (each dial can take up to RouterDialTimeout of 10 seconds). This can cause the reconnect loop to miss ticks or delay connections to other healthy routers.\n\nConsider dialing the unconnected routers concurrently using a sync.WaitGroup to ensure that the entire connection process is bounded by a single dial timeout.
\tvar wg sync.WaitGroup\n\tfor _, r := range routers {\n\t\tinfo, err := routerAddrInfo(r)\n\t\tif err != nil {\n\t\t\tlogger.Warnf(\"[Mesh] Skipping router lease %q: %v\", r.PeerID, err)\n\t\t\tcontinue\n\t\t}\n\t\tif p.host.Network().Connectedness(info.ID) == network.Connected {\n\t\t\tcontinue\n\t\t}\n\t\twg.Add(1)\n\t\tgo func(info peer.AddrInfo) {\n\t\t\tdefer wg.Done()\n\t\t\tdialCtx, cancel := context.WithTimeout(ctx, RouterDialTimeout)\n\t\t\tdefer cancel()\n\t\t\tif err := p.host.Connect(dialCtx, info); err != nil {\n\t\t\t\tlogger.Warnf(\"[Mesh] Router %s unreachable for event publishing: %v\", info.ID, err)\n\t\t\t} else {\n\t\t\t\tlogger.Infof(\"[Mesh] Connected to router %s\", info.ID)\n\t\t\t}\n\t\t}(info)\n\t}\n\twg.Wait()| 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() | ||
| } |
There was a problem hiding this comment.
The Close method is not idempotent. If Close is called multiple times, it will repeatedly invoke p.close(), which closes the underlying libp2p host and pubsub topic multiple times. This can lead to unexpected errors or panics depending on the libp2p implementation.\n\nTo make Close idempotent, set p.close = nil after executing it under the lock.
func (p *P2PMeshAdapter) Close() error {\n\tp.mu.Lock()\n\tdefer p.mu.Unlock()\n\tif p.close == nil {\n\t\treturn nil\n\t}\n\terr := p.close()\n\tp.close = nil\n\treturn err\n}NewServer installs NewNopMeshAdapter and nothing in the binary ever replaced it, so every MeshEvent the code carefully signs (BANNED, KEY_ROTATION, POLICY_UPDATE) was dropped on the floor. Routers and nodes handle those events; they just never received one outside the tests that wire the adapter themselves. The control plane now has a presence on the mesh, and it is as small as it can be: an ephemeral libp2p identity with no listen address, no DHT, no relay and no stream handlers, holding a gossipsub session with the routers so a publish has somewhere to go. It finds the routers the one way the control plane already knows the mesh's shape, the lease table, and re-dials any it lost every --mesh-reconnect-interval. It is one-way: nothing about the mesh is read back, and every consumer keeps pulling /keys, /info and /policies, so a missed event is a delay and never a divergence. sam-one runs the router in the same process, so its control plane publishes on the embedded router's own topic instead of dialing itself. NewP2PMeshAdapter now takes that topic rather than joining one, and Close only tears down what NewMeshPublisher built. Fixes google#317
55cdce0 to
81c0188
Compare
…test The constructor takes the joined topic now; this caller was outside the packages built locally and only CI's typecheck caught it.
What
Fixes #317.
NewServerinstallsNewNopMeshAdapter()and nothing incmd/sam-control-planeever replaced it, so everyMeshEventthe control plane signs (BANNED,KEY_ROTATION,POLICY_UPDATE) was dropped. Routers and nodes handle these events; they just never received one outside the tests that callSetMeshAdapterthemselves.How
The control plane gets a presence on the mesh that is as small as it can be — a publish-only peer:
controlplane.NewMeshPublisher(ctx, store, reconnect): an ephemeral libp2p identity with no listen address, no DHT, no relay, no stream handlers. It can dial routers; nothing can dial it. Gossipsub withStrictSign, joined toapi.GossipEventsbut never subscribed, so routers do not forward traffic back to it.GetActiveRouters). Every--mesh-reconnect-interval(30s, same cadence as the routers'/infopoll) it re-reads the leases and dials any router it is not connected to. Connections are kept warm ahead of events, because a publish only reaches peers whose subscription is already known./keys,/info,/policies(routers today, nodes after node: pull keys, bans and policy from the control plane in one loop #453), so a missed event is a delay and never a divergence. Events are signed by the control plane key as before; the libp2p identity carries no trust.sam-oneruns the router in-process, so its control plane publishes on the embedded router's own topic instead of dialing itself.NewP2PMeshAdapternow takes that topic rather than joining one, andCloseonly tears down whatNewMeshPublisherbuilt.No new dependencies; libp2p and gossipsub were already imported by
internal/controlplane.Why this shape
The alternative — routers polling an event feed on the control plane — needs a persisted event log, cursors, a new proto message and migrations for two databases. This reuses
P2PMeshAdapter(already written and tested) and adds one constructor and one loop. The control plane's coupling to the mesh is: "dial the addresses in the lease table and publish on one topic".Tests
TestMeshPublisherReachesLeasedRouter: a router host subscribed to the topic, known only via a lease (plus an expired lease that must not be dialed); the publisher connects, learns the router's subscription, and aBANNEDevent arrives on the router's subscription. Asserts the publisher listens on nothing and never subscribes.TestRouterAddrInfo: lease addresses with trailing/p2p/<id>become one dial target; an address naming another peer, an undecodable ID, and a lease with no usable address are rejected.TestP2PMeshAdapter_PublishAndSubscribe,TestKeyRotationEventIsSignedByTheRetiringKey,TestCatalogPeerIDCanonicalizationadapted to the topic-taking constructor.Sending to CI for the full suites.