diff --git a/pkg/address-resolver/api/bind_ingest.go b/pkg/address-resolver/api/bind_ingest.go new file mode 100644 index 0000000000..1c596cbede --- /dev/null +++ b/pkg/address-resolver/api/bind_ingest.go @@ -0,0 +1,143 @@ +// Package api pkg/address-resolver/api/bind_ingest.go c4-net-discovery +// +// AR-bind-over-CXO ingest: the server side of the fan-in path where visors +// publish their AR bindings as a CXO feed instead of re-registering over a +// fresh dmsg stream on a timer (each with a full Noise handshake — the +// secp256k1 handshakeResponder that dominates AR CPU). The CXO aggregator +// (pkg/address-resolver/regcxo) calls IngestBindFromCXO for each bind leaf it +// replicates. +// +// This is purely ADDITIVE and dual-write: the HTTP POST /bind and the SUDPH +// UDP registration remain the authoritative path and keep writing the same +// store. The CXO ingest therefore never clobbers a fresh HTTP/UDP bind: +// +// - When a record already exists (the common case — HTTP/UDP bound first), +// the CXO ingest is a keepalive: it re-writes the STORED VisorData to +// refresh its TTL, off a warm CXO connection instead of a fresh handshake. +// It writes back exactly what is there, so it can never overwrite a newer +// HTTP/UDP address (mirrors dmsg-discovery's equal-sequence keepalive, +// which refreshes off the stored entry, not the incoming one). +// - When no record exists yet, only the address-POST types (STCPR/QUIC/WT) +// take a fresh insert, reconstructed from the visor's DECLARED addresses +// exactly as the HTTP bind handler does in the production dmsg-routed case +// (where the observed source is the dmsg bridge and the handler falls back +// to the declared PublicIP). SUDPH is keepalive-only: its stored address is +// the UDP-observed NAT-mapped endpoint, which the declared payload cannot +// reproduce, so a fresh SUDPH record is left to the UDP path. +package api + +import ( + "context" + "net" + + "github.com/skycoin/skywire/pkg/address-resolver/store" + "github.com/skycoin/skywire/pkg/cipher" + "github.com/skycoin/skywire/pkg/netutil" + "github.com/skycoin/skywire/pkg/transport/network/addrresolver" + types "github.com/skycoin/skywire/pkg/transport/types" +) + +// supportedCXOBindType reports whether tpType is a transport type the AR-bind +// CXO feed carries. Anything else is dropped by the aggregator before it +// reaches ingest, but the guard is kept here too so the ingest is safe to call +// directly. +func supportedCXOBindType(tpType types.Type) bool { + switch tpType { + case types.STCPR, types.SUDPH, types.QUIC, types.WT: + return true + default: + return false + } +} + +// IngestBindFromCXO applies an AR binding received over the AR-bind-over-CXO +// feed. reporter is the feed's publisher PK (the visor); the binding is always +// stored UNDER reporter, so a visor can only ever register its OWN address — +// the ownership check the HTTP path enforces via httpauth is structural here. +// +// See the package doc for the dual-write / no-clobber semantics. +func (a *API) IngestBindFromCXO(ctx context.Context, reporter cipher.PubKey, tpType types.Type, la addrresolver.LocalAddresses) { + if reporter == (cipher.PubKey{}) { + return + } + if !supportedCXOBindType(tpType) { + a.log.WithField("reporter", reporter).WithField("type", tpType). + Debug("ar-bind-cxo: unsupported transport type; dropping") + return + } + + // Keepalive-first: if a record already exists, refresh its TTL off the + // STORED data. This is the common case (HTTP/UDP bound first) and the whole + // CPU win — a warm CXO Root keeps the binding alive with no fresh Noise + // handshake. Writing back the stored value can never clobber a newer + // HTTP/UDP bind. + stored, err := a.store.Resolve(ctx, tpType, reporter) + if err == nil { + if bErr := a.store.Bind(ctx, tpType, reporter, stored); bErr != nil { + a.log.WithError(bErr).WithField("reporter", reporter).WithField("type", tpType). + Debug("ar-bind-cxo: TTL keepalive refresh failed") + return + } + a.mirrorForType(tpType, reporter, &stored) + return + } + if err != store.ErrNoEntry && err != store.ErrUnknownTransportType { + a.log.WithError(err).WithField("reporter", reporter).WithField("type", tpType). + Debug("ar-bind-cxo: store lookup failed") + return + } + + // No record yet. SUDPH's stored endpoint is the UDP-observed NAT-mapped + // address, which the declared payload can't reproduce — leave a fresh + // SUDPH record to the authoritative UDP path. + if tpType == types.SUDPH { + return + } + + // Fresh insert for the address-POST types, reconstructed from the declared + // addresses exactly as the HTTP bind handler does when the observed source + // is non-public (the production dmsg-routed default): remoteAddr = declared + // PublicIP. Only when the declaration is a usable public address; otherwise + // leave it to the HTTP path (which can also use the observed source IP). + remoteAddr := la.PublicIP + if !netutil.IsPublicIP(net.ParseIP(remoteAddr)) { + a.log.WithField("reporter", reporter).WithField("type", tpType). + Debug("ar-bind-cxo: no usable declared public IP; deferring fresh bind to HTTP") + return + } + if !a.hasAddress(remoteAddr, la) { + a.log.WithField("reporter", reporter).WithField("type", tpType). + Debug("ar-bind-cxo: declared public IP not in addresses list; dropping") + return + } + v4Addr, v6Addr := splitFamilyAddr(remoteAddr) + if la.PublicIPv6 != "" && isPublicIPv6(net.ParseIP(la.PublicIPv6)) { + v6Addr = la.PublicIPv6 + } + visorData := addrresolver.VisorData{ + RemoteAddr: v4Addr, + RemoteAddrV6: v6Addr, + LocalAddresses: la, + } + if bErr := a.store.Bind(ctx, tpType, reporter, visorData); bErr != nil { + a.log.WithError(bErr).WithField("reporter", reporter).WithField("type", tpType). + Debug("ar-bind-cxo: fresh Bind failed") + return + } + a.mirrorForType(tpType, reporter, &visorData) + a.log.WithField("reporter", reporter).WithField("type", tpType). + Debug("ar-bind-cxo: fresh binding ingested") +} + +// mirrorForType fans the per-transport DHT mirror the HTTP bind path runs +// inline. Only STCPR and SUDPH mirror (matching bindForType / bindSUDPH); +// QUIC/WT have no DHT salt. Best-effort and nil-safe (the mirror methods +// early-return when unconfigured). +func (a *API) mirrorForType(tpType types.Type, pk cipher.PubKey, data *addrresolver.VisorData) { + switch tpType { + case types.STCPR: + a.mirrorSTCPR(pk, data) + case types.SUDPH: + a.mirrorSUDPH(pk, data) + } +} diff --git a/pkg/address-resolver/regcxo/aggregator.go b/pkg/address-resolver/regcxo/aggregator.go new file mode 100644 index 0000000000..70e5c86f5f --- /dev/null +++ b/pkg/address-resolver/regcxo/aggregator.go @@ -0,0 +1,394 @@ +// Package regcxo pkg/address-resolver/regcxo/aggregator.go c4-net-discovery +// +// AR-bind-over-CXO aggregator — the address-resolver side of the fan-in +// path. Visors publish their AR bindings (the stcpr/sudph/quic/wt +// reachable-address payloads they POST to /bind) as a CXO feed and +// AnnounceTo this service; the aggregator owns one CXO Node listening on +// DmsgVisorARBindCXOPort, subscribes to each visor's feed on connect, and +// on every filled Root reads the per-type bind leaves and hands each to +// the Sink (the AR API's IngestBindFromCXO). +// +// This moves address binding off the timer-driven re-registration — each a +// fresh dmsg stream with a full Noise handshake (the secp256k1 +// handshakeResponder that dominates AR CPU) — onto a persistent CXO +// connection kept warm by the treestore heartbeat. The HTTP/UDP bind path +// remains authoritative, so ingest is idempotent (see IngestBindFromCXO). +// +// Structure mirrors pkg/dmsg/discovery/regcxo, widened to several typed +// leaves: same connect-driven subscribe, same grace-gated orphan-feed +// reclaim that keeps the in-memory CXDS from growing without bound as visor +// PKs churn. The node identity is bound to the AR's service SecKey so its +// handshake PK is the AR PK gated visors allowlist (mirrors #4168). +package regcxo + +import ( + "context" + "encoding/json" + "sync" + "time" + + skycipher "github.com/skycoin/skycoin/src/cipher" + + "github.com/skycoin/skywire/pkg/cipher" + "github.com/skycoin/skywire/pkg/cxo/cxoutils" + "github.com/skycoin/skywire/pkg/cxo/node" + cxotransport "github.com/skycoin/skywire/pkg/cxo/node/transport" + "github.com/skycoin/skywire/pkg/cxo/skyobject" + "github.com/skycoin/skywire/pkg/cxo/skyobject/registry" + "github.com/skycoin/skywire/pkg/cxo/treestore" + "github.com/skycoin/skywire/pkg/dmsg/dmsg" + "github.com/skycoin/skywire/pkg/logging" + "github.com/skycoin/skywire/pkg/skyenv" + "github.com/skycoin/skywire/pkg/transport/network/addrresolver" + types "github.com/skycoin/skywire/pkg/transport/types" +) + +// cxoBindLeaf maps a CXO leaf name to the transport type it carries. The leaf +// name is the type's canonical wire string, matching what the visor's AR-bind +// publisher Puts (see addrresolver bind hooks + pkg/visor/init_ar_bind_cxo.go). +type cxoBindLeaf struct { + name string + t types.Type +} + +// cxoBindLeaves is the fixed set of per-type leaves the AR-bind feed carries. +var cxoBindLeaves = []cxoBindLeaf{ + {"stcpr", types.STCPR}, + {"sudph", types.SUDPH}, + {"squicr", types.QUIC}, + {"swtr", types.WT}, +} + +// Sink ingests bindings replicated from visor AR-bind feeds. The AR API +// satisfies it via (*api.API).IngestBindFromCXO. +type Sink interface { + IngestBindFromCXO(ctx context.Context, reporter cipher.PubKey, tpType types.Type, la addrresolver.LocalAddresses) +} + +// Config tunes the aggregator loops. Zero values get sane defaults. +type Config struct { + ReconcileInterval time.Duration + CleanupInterval time.Duration + MaxFillingTime time.Duration + Logger *logging.Logger + InMemoryDB bool + DataDir string + // SecKey binds the aggregator's CXO node identity to the AR's service + // secret key so the node's handshake-advertised PK is the AR's KNOWN PK. + // This matters because a visor gates its feed subscriber allowlist on the + // CXO node's PeerID: it allows the AR PK it holds in + // transport.address_resolver_dmsg. Left zero, node.NewNode generates a + // RANDOM keypair, so the aggregator dials every gated visor as an unknown + // PK and is rejected (the #4168 bug on TPD's aggregator). The publisher + // path (treestore.NewWithDMSG) binds the same way for the same reason. + SecKey cipher.SecKey +} + +// orphanGraceTicks is how many consecutive cleanup ticks a feed must have +// zero connected conns before it is reclaimed. At the 2-minute default +// CleanupInterval this is a ~4-minute grace — long enough to ride out a +// visor's dmsg reconnect without churning a stable feed. +const orphanGraceTicks = 2 + +// Aggregator owns one CXO Node listening on DMSG; visors dial in, subscribe +// happens per-conn during reconcile, and OnRootFilled reads the bind leaves +// and forwards them to the Sink. +type Aggregator struct { + cxoNode *node.Node + sink Sink + conf Config + log *logging.Logger + + mu sync.Mutex + cancel context.CancelFunc + done chan struct{} + + // nudge triggers an immediate reconcile out of band from the ticker. + // Buffered(1) so a burst of connects coalesces into one pending reconcile + // (idempotent — it walks the full conn set anyway). + nudge chan struct{} + + // orphanStrikes counts consecutive cleanup ticks a feed has had no + // connected conn. Touched only from cleanup() (single goroutine). + orphanStrikes map[skycipher.PubKey]int +} + +// New constructs an Aggregator: a CXO Node with DMSG enabled on +// DmsgVisorARBindCXOPort so remote visors can dial in, wired to forward each +// filled Root's bind leaves to sink. +func New(dmsgC *dmsg.Client, sink Sink, conf Config) (*Aggregator, error) { + if conf.ReconcileInterval <= 0 { + conf.ReconcileInterval = 30 * time.Second + } + if conf.CleanupInterval <= 0 { + conf.CleanupInterval = 2 * time.Minute + } + if conf.MaxFillingTime <= 0 { + conf.MaxFillingTime = 90 * time.Second + } + if conf.Logger == nil { + conf.Logger = logging.MustGetLogger("ar-bind-cxo") + } + + cfg := node.NewConfig() + cfg.MaxFillingTime = conf.MaxFillingTime + cfg.Config = skyobject.NewConfig() + cfg.Config.InMemoryDB = conf.InMemoryDB || conf.DataDir == "" + if conf.DataDir != "" { + cfg.Config.DataDir = conf.DataDir + } + // Bind the node identity to the AR's service key (when provided) so its + // handshake PK is the AR's known PK, matching what gated visors allowlist. + // Zero SecKey => node.NewNode mints a random keypair => gated visors reject + // the aggregator's subscribe. See Config.SecKey. + if conf.SecKey != (cipher.SecKey{}) { + cfg.SecKey = skycipher.SecKey(conf.SecKey) + } + // We're DMSG-only — disable the CXO node's default TCP/RPC listeners. + // node.NewConfig defaults TCP.Listen to ":8870" and RPC to ":8871"; those + // hardcoded ports would collide with any other CXO node in the same process + // (e.g. when the multi-service supervisor runs AR alongside another CXO + // service). DMSG is enabled separately below. + cfg.TCP.Listen = "" + cfg.RPC = "" + + cxoNode, err := node.NewNode(cfg) + if err != nil { + return nil, err + } + factory := cxotransport.NewDMSGFactory(dmsgC, skyenv.DmsgVisorARBindCXOPort) + if err := cxoNode.EnableDMSG(factory); err != nil { + _ = cxoNode.Close() //nolint:errcheck + return nil, err + } + + a := &Aggregator{ + cxoNode: cxoNode, + sink: sink, + conf: conf, + log: conf.Logger, + done: make(chan struct{}), + nudge: make(chan struct{}, 1), + orphanStrikes: make(map[skycipher.PubKey]int), + } + + // Subscribe the moment a visor dials in, rather than waiting up to a full + // ReconcileInterval. The conn's handshake completes (peerID set) before + // OnConnect fires, so the nudged reconcile can subscribe at once; it stays + // idempotent via alreadySubscribed. + cxoNode.Config().OnConnect = func(_ *node.Conn) error { + select { + case a.nudge <- struct{}{}: + default: + } + return nil + } + cxoNode.Config().OnRootFilled = func(_ *node.Node, r *registry.Root) { + a.handleRootFilled(r) + } + cxoNode.Config().OnFillingBreaks = func(_ *node.Node, r *registry.Root, reason error) { + a.log.WithError(reason).WithField("visor", cipher.PubKey(r.Pub)). + Debug("ar-bind-cxo aggregator: root filling broke") + } + return a, nil +} + +// Run starts the reconcile + cleanup loops. Returns immediately; the loop runs +// until ctx is canceled or Close is called. Idempotent. +func (a *Aggregator) Run(ctx context.Context) { + a.mu.Lock() + if a.cancel != nil { + a.mu.Unlock() + return + } + loopCtx, cancel := context.WithCancel(ctx) + a.cancel = cancel + a.mu.Unlock() + go a.loop(loopCtx) +} + +// Close stops the loops and tears down the CXO node. Idempotent. +func (a *Aggregator) Close() error { + a.mu.Lock() + cancel := a.cancel + a.cancel = nil + a.mu.Unlock() + if cancel != nil { + cancel() + <-a.done + } + return a.cxoNode.Close() +} + +// FeedPK returns the aggregator's own CXO node identity (the AR PK when SecKey +// is bound). +func (a *Aggregator) FeedPK() cipher.PubKey { return cipher.PubKey(a.cxoNode.ID()) } + +func (a *Aggregator) loop(ctx context.Context) { + defer close(a.done) + t := time.NewTicker(a.conf.ReconcileInterval) + defer t.Stop() + ct := time.NewTicker(a.conf.CleanupInterval) + defer ct.Stop() + + a.reconcile() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + a.reconcile() + case <-ct.C: + a.cleanup() + case <-a.nudge: + a.reconcile() + } + } +} + +// reconcile subscribes to every connected visor's own feed (feed PK == peer +// PK) that isn't already subscribed. Dropped conns simply aren't in the list +// anymore. +func (a *Aggregator) reconcile() { + for _, conn := range a.cxoNode.Connections() { + peerPK := conn.PeerID() + if peerPK == (skycipher.PubKey{}) { + continue + } + if alreadySubscribed(conn, peerPK) { + continue + } + if err := conn.Subscribe(peerPK); err != nil { + a.log.WithError(err).WithField("visor", cipher.PubKey(peerPK)). + Debug("ar-bind-cxo aggregator: Subscribe failed; will retry next reconcile") + continue + } + a.log.WithField("visor", cipher.PubKey(peerPK)). + Debug("ar-bind-cxo aggregator: subscribed to visor feed") + } +} + +func alreadySubscribed(conn *node.Conn, feed skycipher.PubKey) bool { + for _, f := range conn.Feeds() { + if f == feed { + return true + } + } + return false +} + +// cleanup prunes superseded Roots (keeping only the latest per feed) and +// reclaims feeds whose visor has gone away, then sweeps ownerless objects. +// Without this the in-memory store grows without bound as visor PKs churn. +func (a *Aggregator) cleanup() { + c := a.cxoNode.Container() + if err := cxoutils.RemoveRootObjects(c, 1); err != nil { + a.log.WithError(err).Debug("ar-bind-cxo aggregator: RemoveRootObjects failed; will retry next tick") + return + } + a.reclaimOrphanFeeds(c) + if err := cxoutils.RemoveObjects(c); err != nil { + a.log.WithError(err).Debug("ar-bind-cxo aggregator: RemoveObjects failed; will retry next tick") + } +} + +// reclaimOrphanFeeds un-shares and hard-deletes feeds that no longer have a +// connected conn, grace-gated by orphanGraceTicks so a brief drop + redial +// keeps the feed. +func (a *Aggregator) reclaimOrphanFeeds(c *skyobject.Container) { + connected := make(map[skycipher.PubKey]struct{}) + for _, conn := range a.cxoNode.Connections() { + if pk := conn.PeerID(); pk != (skycipher.PubKey{}) { + connected[pk] = struct{}{} + } + } + self := a.cxoNode.ID() + for _, feed := range c.Feeds() { + if feed == self { + delete(a.orphanStrikes, feed) + continue + } + if _, ok := connected[feed]; ok { + delete(a.orphanStrikes, feed) + continue + } + a.orphanStrikes[feed]++ + if a.orphanStrikes[feed] < orphanGraceTicks { + continue + } + delete(a.orphanStrikes, feed) + if err := a.cxoNode.DontShare(feed); err != nil { + a.log.WithError(err).WithField("visor", cipher.PubKey(feed)). + Debug("ar-bind-cxo aggregator: DontShare orphan feed failed; will retry next tick") + continue + } + if err := c.DelFeed(feed); err != nil { + a.log.WithError(err).WithField("visor", cipher.PubKey(feed)). + Debug("ar-bind-cxo aggregator: DelFeed orphan feed failed; will retry next tick") + continue + } + a.log.WithField("visor", cipher.PubKey(feed)). + Debug("ar-bind-cxo aggregator: reclaimed orphan feed (no connected conn)") + } +} + +// handleRootFilled reads every per-type bind leaf from a filled Root and +// forwards each decoded LocalAddresses to the Sink. r.Pub is the visor whose +// feed produced this Root — the reporter PK the ingest stores the binding +// under. +func (a *Aggregator) handleRootFilled(r *registry.Root) { + if r == nil || len(r.Refs) == 0 { + return + } + reporter := cipher.PubKey(r.Pub) + if reporter == (cipher.PubKey{}) { + a.log.Debug("ar-bind-cxo aggregator: dropping root with zero publisher PK") + return + } + pack, err := a.cxoNode.Container().Pack(r, treestore.Registry) + if err != nil { + a.log.WithError(err).Debug("ar-bind-cxo aggregator: get pack failed") + return + } + var rootNode treestore.TreeNode + if err := r.Refs[0].Value(pack, &rootNode); err != nil { + a.log.WithError(err).Debug("ar-bind-cxo aggregator: decode root TreeNode failed") + return + } + + for _, bl := range cxoBindLeaves { + leaf, ok := findLeaf(pack, &rootNode, bl.name) + if !ok || len(leaf) == 0 { + continue + } + var la addrresolver.LocalAddresses + if err := json.Unmarshal(leaf, &la); err != nil { + a.log.WithError(err).WithField("visor", reporter).WithField("type", bl.t). + Debug("ar-bind-cxo aggregator: bind leaf decode failed") + continue + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + a.sink.IngestBindFromCXO(ctx, reporter, bl.t, la) + cancel() + } +} + +// findLeaf returns the leaf value stored directly under n at the given +// top-level name. The AR-bind feed is flat (one leaf per type), so a single +// level of lookup suffices — no recursive walk. +func findLeaf(pack registry.Pack, n *treestore.TreeNode, name string) ([]byte, bool) { + count, err := n.Children.Len(pack) + if err != nil { + return nil, false + } + for i := 0; i < count; i++ { + var entry treestore.TreeEntry + if _, err := n.Children.ValueByIndex(pack, i, &entry); err != nil { + continue + } + if entry.Name == name && len(entry.Leaf) > 0 { + return entry.Leaf, true + } + } + return nil, false +} diff --git a/pkg/services/ar/ar.go b/pkg/services/ar/ar.go index 2203efcc1e..34c954f2be 100644 --- a/pkg/services/ar/ar.go +++ b/pkg/services/ar/ar.go @@ -14,12 +14,14 @@ import ( "github.com/skycoin/skywire/deployment" "github.com/skycoin/skywire/pkg/address-resolver/api" armetrics "github.com/skycoin/skywire/pkg/address-resolver/metrics" + "github.com/skycoin/skywire/pkg/address-resolver/regcxo" "github.com/skycoin/skywire/pkg/address-resolver/store" "github.com/skycoin/skywire/pkg/dmsg/dmsg" "github.com/skycoin/skywire/pkg/httpauth" "github.com/skycoin/skywire/pkg/logging" "github.com/skycoin/skywire/pkg/metricsutil" "github.com/skycoin/skywire/pkg/services" + "github.com/skycoin/skywire/pkg/skyenv" "github.com/skycoin/skywire/pkg/storeconfig" "github.com/skycoin/skywire/pkg/svcmode" ) @@ -199,6 +201,27 @@ func (s *service) Run(ctx context.Context) error { } defer h.Close() + // AR-bind-over-CXO aggregator: always-on fan-in path where visors publish + // their AR bindings as a CXO feed instead of re-registering over a fresh + // dmsg stream (each a full Noise handshake) on a timer. Inert until visors + // subscribe (just a listener), purely additive to the authoritative + // HTTP/UDP bind path, so it needs no gate. Needs the dmsg client; the API + // is the Sink (IngestBindFromCXO). The node identity is bound to the AR's + // service SecKey so gated visors accept its subscribe (see #4168). + // Best-effort — HTTP/UDP registration is unaffected if it fails to start. + if h.DmsgClient != nil { + agg, aerr := regcxo.New(h.DmsgClient, arAPI, regcxo.Config{SecKey: sk, Logger: logger}) + if aerr != nil { + logger.WithError(aerr).Error("Failed to start AR-bind-over-CXO aggregator, continuing without it") + } else { + agg.Run(runCtx) + defer func() { _ = agg.Close() }() //nolint:errcheck + logger.WithField("feed_pk", agg.FeedPK()). + WithField("port", skyenv.DmsgVisorARBindCXOPort). + Info("AR-bind-over-CXO aggregator running") + } + } + defer arAPI.Close() select { case <-runCtx.Done(): diff --git a/pkg/transport/network/addrresolver/client.go b/pkg/transport/network/addrresolver/client.go index 10e3c0c052..b85f976cf3 100644 --- a/pkg/transport/network/addrresolver/client.go +++ b/pkg/transport/network/addrresolver/client.go @@ -157,6 +157,46 @@ type httpClient struct { ready chan struct{} closed chan struct{} delBindSudphWg sync.WaitGroup + + // bindPublishHook, when set via SetBindPublishHook, is invoked with the + // canonical transport-type wire name and the exact LocalAddresses payload + // on every successful Bind{STCPR,QUIC,WT} and SUDPH (re)registration. It + // lets the visor mirror its AR bindings onto a CXO feed + // (init_ar_bind_cxo.go) without this package depending on treestore/CXO. + // Guarded by bindHookMu; never blocks the bind path (the hook coalesces + // into the publisher's batch window). + bindHookMu sync.RWMutex + bindPublishHook func(netType string, payload LocalAddresses) +} + +// BindPublisher is the optional extension a caller type-asserts the APIClient +// to in order to mirror AR bindings onto a side channel (the CXO AR-bind +// feed). Only the httpClient implements it; keeping it off APIClient avoids +// touching the mock and the TinyGo-facing client variants (mirrors how +// SUDPHBinder is an optional extension). +type BindPublisher interface { + // SetBindPublishHook installs fn (or clears it with nil). fn is called + // with the canonical transport-type wire name ("stcpr", "sudph", + // "squicr", "swtr") and the LocalAddresses the visor just registered. + SetBindPublishHook(fn func(netType string, payload LocalAddresses)) +} + +// SetBindPublishHook implements BindPublisher. +func (c *httpClient) SetBindPublishHook(fn func(netType string, payload LocalAddresses)) { + c.bindHookMu.Lock() + c.bindPublishHook = fn + c.bindHookMu.Unlock() +} + +// fireBindPublishHook invokes the installed bind-publish hook, if any. Safe to +// call with the hook unset. Never panics on a nil hook. +func (c *httpClient) fireBindPublishHook(netType string, payload LocalAddresses) { + c.bindHookMu.RLock() + hook := c.bindPublishHook + c.bindHookMu.RUnlock() + if hook != nil { + hook(netType, payload) + } } // NewHTTP creates a new client setting a public key to the client to be used for auth. @@ -544,6 +584,8 @@ func (c *httpClient) BindQUIC(ctx context.Context, port string) error { if resp.StatusCode != http.StatusOK { return fmt.Errorf("status: %d, error: %w", resp.StatusCode, httpauthclient.ExtractError(resp.Body)) } + // Mirror onto the CXO AR-bind feed. Wire name matches types.QUIC. + c.fireBindPublishHook("squicr", localAddresses) return nil } @@ -607,6 +649,8 @@ func (c *httpClient) BindWT(ctx context.Context, port, certHash string) error { if resp.StatusCode != http.StatusOK { return fmt.Errorf("status: %d, error: %w", resp.StatusCode, httpauthclient.ExtractError(resp.Body)) } + // Mirror onto the CXO AR-bind feed. Wire name matches types.WT. + c.fireBindPublishHook("swtr", localAddresses) return nil } @@ -679,6 +723,10 @@ func (c *httpClient) BindSTCPR(ctx context.Context, port string) error { return fmt.Errorf("status: %d, error: %w", resp.StatusCode, httpauthclient.ExtractError(resp.Body)) } + // Mirror the just-registered binding onto the CXO AR-bind feed (when a + // publisher installed a hook). Wire name matches types.STCPR. + c.fireBindPublishHook("stcpr", localAddresses) + // #1525 Phase 2b: when v6 is available, fire a SECONDARY POST over // the v6-forced auth client. The AR's bind handler captures the // connecting socket's family via splitFamilyAddr (#2715 Phase 1) diff --git a/pkg/transport/network/addrresolver/client_sudph.go b/pkg/transport/network/addrresolver/client_sudph.go index 87f247b004..cefe518472 100644 --- a/pkg/transport/network/addrresolver/client_sudph.go +++ b/pkg/transport/network/addrresolver/client_sudph.go @@ -138,6 +138,13 @@ func (c *httpClient) connectSUDPH(filter *pfilter.PacketFilter, hs Handshake) (n return nil, LocalAddresses{}, err } + // Mirror the SUDPH registration onto the CXO AR-bind feed (when a + // publisher installed a hook). Wire name matches types.SUDPH. The AR's + // CXO ingest treats SUDPH as keepalive-only (it cannot reproduce the + // UDP-observed NAT-mapped address), so this refreshes the store TTL + // without a fresh dmsg/Noise handshake on every re-registration. + c.fireBindPublishHook("sudph", localAddresses) + return arConn, localAddresses, nil } diff --git a/pkg/visor/init.go b/pkg/visor/init.go index a67d3c0d74..e512cf15ab 100644 --- a/pkg/visor/init.go +++ b/pkg/visor/init.go @@ -154,6 +154,9 @@ var ( // regCXOMod publishes this visor's discovery entry as a CXO feed // (registration-over-CXO) when opted in regCXOMod vinit.Module + // arBindCXOMod mirrors this visor's AR bindings onto a CXO feed the + // address-resolver aggregates (AR-bind-over-CXO), always-on/additive + arBindCXOMod vinit.Module // visor that groups all modules together vis vinit.Module ) @@ -267,8 +270,14 @@ func registerModules(logger *logging.MasterLogger) { // dmsgC (the entry it publishes + the feed's transport). See // init_registration_cxo.go. regCXOMod = maker("registration_cxo", initRegistrationCXO, &dmsgC) + // AR-bind-over-CXO publisher: mirror this visor's address-resolver + // bindings onto a CXO feed the AR aggregates, off the timer-driven + // re-registration (each a fresh dmsg Noise handshake). Depends on the AR + // client (the bind hook it publishes from) and dmsgC (the feed transport). + // See init_ar_bind_cxo.go. + arBindCXOMod = maker("ar_bind_cxo", initARBindCXO, &ar, &dmsgC) vis = vinit.MakeModule("visor", vinit.DoNothing, logger, &ebc, &ar, &disc, &ptyModule, - &tr, &rt, &launch, &cli, &hvs, &ut, &pv, &pvs, &trs, &stcpC, &stcprC, &quicC, &wsC, &wtC, &skyFwd, &pi, &dmsgPi, &dmsgSrv, &dmsgServerLatency, &systemSurvey, &tc, &tpdco, &embTPS, &embRouteSetup, &embDmsgWeb, &embFwdProxy, &embSkynetWeb, &meshProxy, &embSkymailBridge, &uiServer, &nodeHealth, &selfProbe, &skynetPorts, &statsMod, &cxoUserFeedsMod, &pairingMod, &groupingMod, &voiceMod, &coinNodesMod, ®CXOMod) + &tr, &rt, &launch, &cli, &hvs, &ut, &pv, &pvs, &trs, &stcpC, &stcprC, &quicC, &wsC, &wtC, &skyFwd, &pi, &dmsgPi, &dmsgSrv, &dmsgServerLatency, &systemSurvey, &tc, &tpdco, &embTPS, &embRouteSetup, &embDmsgWeb, &embFwdProxy, &embSkynetWeb, &meshProxy, &embSkymailBridge, &uiServer, &nodeHealth, &selfProbe, &skynetPorts, &statsMod, &cxoUserFeedsMod, &pairingMod, &groupingMod, &voiceMod, &coinNodesMod, ®CXOMod, &arBindCXOMod) // Hypervisor includes the full visor module tree so all services // (CLI, transports, pings, public visor, etc.) run in hypervisor mode. diff --git a/pkg/visor/init_ar_bind_cxo.go b/pkg/visor/init_ar_bind_cxo.go new file mode 100644 index 0000000000..e4cadc1933 --- /dev/null +++ b/pkg/visor/init_ar_bind_cxo.go @@ -0,0 +1,158 @@ +// Package visor pkg/visor/init_ar_bind_cxo.go c3-vis-core +// +// AR-bind-over-CXO publisher. The visor mirrors its address-resolver +// bindings — the same stcpr/sudph/quic/wt reachable-address payloads it +// POSTs to /bind (and re-registers over a fresh dmsg stream every ~90s) — +// onto a persistent CXO feed on DmsgVisorARBindCXOPort and announces to +// the AR, which subscribes back and ingests them. This moves the address +// binding off the timer-driven re-registration, each a fresh dmsg stream +// with a full Noise handshake (the secp256k1 ECDH handshakeResponder that +// dominates AR CPU), onto one warm CXO connection. +// +// Purely ADDITIVE dual-write: the AR client keeps doing the HTTP POST / +// UDP registration exactly as before (the authoritative/fallback path). +// The publisher is fed by a hook the AR client fires on every successful +// bind (see addrresolver.BindPublisher), so the CXO leaf always carries +// the exact LocalAddresses the visor last registered. It is inert until an +// AR subscribes to the feed. +package visor + +import ( + "context" + "encoding/json" + "path/filepath" + "sync/atomic" + "time" + + "github.com/skycoin/skywire/pkg/cipher" + "github.com/skycoin/skywire/pkg/cxo/treestore" + "github.com/skycoin/skywire/pkg/logging" + "github.com/skycoin/skywire/pkg/skyenv" + "github.com/skycoin/skywire/pkg/transport/network/addrresolver" +) + +// arBindBatchWindow coalesces bind mutations into the CXO datastore. Binds +// change rarely (only when a reachable address changes) and the SUDPH +// re-registration is a periodic keepalive, so a short window adds no latency +// while batching the startup burst of per-type binds into one publish. +const arBindBatchWindow = 5 * time.Second + +// arBindAnnounceInterval is how often the visor pokes its CXO feed conn to the +// AR. Short enough that a recovered visor/AR re-links within the health window. +const arBindAnnounceInterval = 30 * time.Second + +func initARBindCXO(_ context.Context, v *Visor, log *logging.Logger) error { + if v.dmsgC == nil { + log.Warn("AR-bind-CXO: dmsg client absent; publisher not started") + return nil + } + // The whole point is to announce to the AR so it subscribes back. Without + // a dmsg:// AR PK there is no announce target, so skip rather than run an + // orphan publisher. + arPK, ok := arBindCXOPeer(v) + if !ok { + log.Warn("AR-bind-CXO: no dmsg:// address_resolver PK; cannot announce to AR; skipping") + return nil + } + + // The publisher is driven by the AR client's bind hook. If the AR client + // isn't the hook-capable http client (or hasn't been constructed), there is + // nothing to mirror — skip. arBindCXOMod depends on the address_resolver + // module, so v.arClient is normally set by the time we run. + v.initLock.Lock() + arClient := v.arClient + v.initLock.Unlock() + bp, ok := arClient.(addrresolver.BindPublisher) + if !ok { + log.Warn("AR-bind-CXO: AR client does not support bind publish hook; skipping") + return nil + } + + dataDir := filepath.Join(v.conf.LocalPath, "cxo-ar-bind") + pub, err := treestore.NewWithDMSG(v.dmsgC, v.conf.SK, treestore.PubConfig{ + DmsgPort: skyenv.DmsgVisorARBindCXOPort, + BatchWindow: arBindBatchWindow, + Logger: log, + DataDir: dataDir, + // Each leaf is rebuilt from the AR client's live bind hook on every + // restart (the AR re-registers on boot), so skipping per-tx fdatasync + // is safe (matches the registration + telemetry publishers). + NoSyncCXDS: true, + }) + if err != nil { + log.WithError(err).Warn("AR-bind-CXO: publisher init failed; continuing with HTTP/UDP AR registration only") + return nil + } + + // Mirror every successful AR bind onto the feed, one leaf per transport + // type (leaf name == the type's canonical wire string). The hook runs on + // the AR client's bind goroutine and must not block; Put coalesces into the + // next BatchWindow tick, so it returns at once. + bp.SetBindPublishHook(func(netType string, payload addrresolver.LocalAddresses) { + b, mErr := json.Marshal(payload) + if mErr != nil { + log.WithError(mErr).Debug("AR-bind-CXO: marshal bind payload failed") + return + } + if pErr := pub.Put(netType, b); pErr != nil { + log.WithError(pErr).Debug("AR-bind-CXO: CXO Put failed") + } + }) + + // Dial the AR so its aggregator sees the inbound conn and subscribes to our + // feed (PK = our PK). ConnectPK is idempotent, so the same loop handles + // initial announce + reconnect. + lastAnnounceOK := new(atomic.Int64) + go runARBindAnnounceLoop(v.ctx, pub, arPK, lastAnnounceOK, log) + + v.pushCloseStack("ar_bind_cxo", func() error { + bp.SetBindPublishHook(nil) + return pub.Close() + }) + + log.WithField("feed_pk", pub.Feed()).WithField("ar_pk", arPK). + WithField("port", skyenv.DmsgVisorARBindCXOPort). + Info("AR-bind-CXO: publisher running (bindings mirrored to address-resolver)") + return nil +} + +// arBindCXOPeer extracts the AR's dmsg publisher PK from the visor's transport +// config, preferring the explicit AddressResolverDmsg URL and falling back to +// AddressResolver (the dmsg-only default stores the dmsg:// URL there). Mirrors +// initAddressResolver's URL selection. +func arBindCXOPeer(v *Visor) (cipher.PubKey, bool) { + if v.conf.Transport == nil { + return cipher.PubKey{}, false + } + raw := v.conf.Transport.AddressResolverDmsg + if raw == "" { + raw = v.conf.Transport.AddressResolver + } + return parseDmsgPeer(raw) +} + +// runARBindAnnounceLoop dials the AR on a ticker (ConnectPK is idempotent — a +// live conn is a no-op, a dropped one redials) and stamps lastOK with the +// wall-clock time of each success. +func runARBindAnnounceLoop(ctx context.Context, pub *treestore.Publisher, arPK cipher.PubKey, lastOK *atomic.Int64, log *logging.Logger) { + t := time.NewTicker(arBindAnnounceInterval) + defer t.Stop() + announce := func() { + dctx, cancel := context.WithTimeout(ctx, arBindAnnounceInterval/2) + defer cancel() + if err := pub.AnnounceTo(dctx, arPK); err != nil { + log.WithError(err).WithField("ar_pk", arPK).Trace("AR-bind-CXO: announce to AR failed") + return + } + lastOK.Store(time.Now().UnixNano()) + } + announce() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + announce() + } + } +}