ziti/controller/network/network.go

/*
	Copyright NetFoundry Inc.

	Licensed under the Apache License, Version 2.0 (the "License");
	you may not use this file except in compliance with the License.
	You may obtain a copy of the License at

	https://www.apache.org/licenses/LICENSE-2.0

	Unless required by applicable law or agreed to in writing, software
	distributed under the License is distributed on an "AS IS" BASIS,
	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	See the License for the specific language governing permissions and
	limitations under the License.
*/

package network

import (
	"bytes"
	"compress/gzip"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"os"
	"runtime/debug"
	"slices"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/michaelquigley/pfxlog"
	"github.com/openziti/channel/v5"
	"github.com/openziti/channel/v5/protobufs"
	"github.com/openziti/foundation/v2/concurrenz"
	"github.com/openziti/foundation/v2/debugz"
	"github.com/openziti/foundation/v2/goroutines"
	"github.com/openziti/foundation/v2/versions"
	"github.com/openziti/identity"
	"github.com/openziti/metrics"
	"github.com/openziti/ziti/v2/common/ctrl_msg"
	"github.com/openziti/ziti/v2/common/inspect"
	"github.com/openziti/ziti/v2/common/logcontext"
	"github.com/openziti/ziti/v2/common/pb/cmd_pb"
	"github.com/openziti/ziti/v2/common/pb/ctrl_pb"
	"github.com/openziti/ziti/v2/common/pb/mgmt_pb"
	"github.com/openziti/ziti/v2/common/servermetrics"
	"github.com/openziti/ziti/v2/common/servermetrics/metrics_pb"
	"github.com/openziti/ziti/v2/common/trace"
	"github.com/openziti/ziti/v2/controller/command"
	"github.com/openziti/ziti/v2/controller/config"
	"github.com/openziti/ziti/v2/controller/db"
	"github.com/openziti/ziti/v2/controller/event"
	"github.com/openziti/ziti/v2/controller/idgen"
	"github.com/openziti/ziti/v2/controller/model"
	"github.com/openziti/ziti/v2/controller/raft"
	"github.com/openziti/ziti/v2/controller/storage/boltz"
	"github.com/openziti/ziti/v2/controller/storage/objectz"
	"github.com/openziti/ziti/v2/controller/xt"
	"github.com/sirupsen/logrus"
	"github.com/teris-io/shortid"
	"go.etcd.io/bbolt"
	"google.golang.org/protobuf/proto"
)

const SmartRerouteAttempt = 99969996

// Config provides the values needed to create a Network instance
type Config interface {
	GetId() *identity.TokenId
	GetMetricsRegistry() metrics.Registry
	GetOptions() *config.NetworkConfig
	GetCommandDispatcher() command.Dispatcher
	GetDb() boltz.Db
	GetVersionProvider() versions.VersionProvider
	GetEventDispatcher() event.Dispatcher
	GetCloseNotify() <-chan struct{}
	IsRaftEnabled() bool
	IsRaftLeader() bool
}

type InspectTarget func(string) (bool, *string, error)

// CtrlDialerValidator is a function that validates ctrl dialer states and returns per-router details.
type CtrlDialerValidator func() ([]*mgmt_pb.ControllerDialerDetails, error)

type Network struct {
	*model.Managers
	env                    model.Env
	nodeId                 string
	options                *config.NetworkConfig
	routeSenderController  *routeSenderController
	eventDispatcher        event.Dispatcher
	traceController        trace.Controller
	routerPresenceHandlers concurrenz.CopyOnWriteSlice[model.RouterPresenceHandler]
	capabilities           []string
	closeNotify            <-chan struct{}
	watchdogCh             chan struct{}
	lock                   sync.Mutex
	strategyRegistry       xt.Registry
	lastSnapshot           time.Time
	metricsRegistry        metrics.Registry
	VersionProvider        versions.VersionProvider

	serviceEventMetrics          servermetrics.UsageRegistry
	serviceDialSuccessCounter    servermetrics.IntervalCounter
	serviceDialFailCounter       servermetrics.IntervalCounter
	serviceDialTimeoutCounter    servermetrics.IntervalCounter
	serviceDialOtherErrorCounter servermetrics.IntervalCounter

	serviceTerminatorTimeoutCounter           servermetrics.IntervalCounter
	serviceTerminatorConnectionRefusedCounter servermetrics.IntervalCounter
	serviceInvalidTerminatorCounter           servermetrics.IntervalCounter
	serviceMisconfiguredTerminatorCounter     servermetrics.IntervalCounter

	config Config

	// restartSelfOnSnapshot, when true, restarts this controller in place after a snapshot restore
	// (e.g. a node joining the cluster) instead of exiting and relying on an external process manager.
	restartSelfOnSnapshot bool

	Inspections      *InspectionsManager
	RouterMessaging  *RouterMessaging
	routerConnectSem concurrenz.Semaphore
	// routerConnectWaiting counts goroutines parked on routerConnectSem. Reported as a gauge because it
	// is the only one of these signals visible during a hang: the timers record on completion, so a
	// wedged setup makes them go quiet exactly when something is wrong.
	routerConnectWaiting atomic.Int64
	// Resolved once and held. A Meter lookup takes a reference each time it is called, and nothing
	// releases them, so looking one up per event leaks a reference per event. The timers do not, but are
	// held alongside so the call site does not invite the question, and so converting one to a meter
	// later cannot quietly reintroduce the leak.
	routerConnectWaitTimer metrics.Timer
	routerConnectWorkTimer metrics.Timer
	routerConnectTimeouts  metrics.Meter
	inspectionTargets      concurrenz.CopyOnWriteSlice[InspectTarget]
	ctrlDialerValidator    CtrlDialerValidator
}

func NewNetwork(config Config, env model.Env) (*Network, error) {
	metricsConfig := servermetrics.DefaultUsageRegistryConfig(config.GetId().Token, config.GetCloseNotify())
	if config.GetOptions().IntervalAgeThreshold != 0 {
		metricsConfig.IntervalAgeThreshold = config.GetOptions().IntervalAgeThreshold
		logrus.Infof("set interval age threshold to '%v'", config.GetOptions().IntervalAgeThreshold)
	}
	serviceEventMetrics := servermetrics.NewUsageRegistry(metricsConfig)

	network := &Network{
		env:                   env,
		Managers:              env.GetManagers(),
		nodeId:                config.GetId().Token,
		options:               config.GetOptions(),
		routeSenderController: newRouteSenderController(),
		eventDispatcher:       config.GetEventDispatcher(),
		traceController:       trace.NewController(config.GetCloseNotify()),
		closeNotify:           config.GetCloseNotify(),
		watchdogCh:            make(chan struct{}, 1),
		strategyRegistry:      xt.GlobalRegistry(),
		lastSnapshot:          time.Now().Add(-time.Hour),
		metricsRegistry:       config.GetMetricsRegistry(),
		VersionProvider:       config.GetVersionProvider(),

		serviceEventMetrics:          serviceEventMetrics,
		serviceDialSuccessCounter:    serviceEventMetrics.IntervalCounter("service.dial.success", time.Minute),
		serviceDialFailCounter:       serviceEventMetrics.IntervalCounter("service.dial.fail", time.Minute),
		serviceDialTimeoutCounter:    serviceEventMetrics.IntervalCounter("service.dial.timeout", time.Minute),
		serviceDialOtherErrorCounter: serviceEventMetrics.IntervalCounter("service.dial.error_other", time.Minute),

		serviceTerminatorTimeoutCounter:           serviceEventMetrics.IntervalCounter("service.dial.terminator.timeout", time.Minute),
		serviceTerminatorConnectionRefusedCounter: serviceEventMetrics.IntervalCounter("service.dial.terminator.connection_refused", time.Minute),
		serviceInvalidTerminatorCounter:           serviceEventMetrics.IntervalCounter("service.dial.terminator.invalid", time.Minute),
		serviceMisconfiguredTerminatorCounter:     serviceEventMetrics.IntervalCounter("service.dial.terminator.misconfigured", time.Minute),

		config: config,
	}

	env.GetManagers().Command.Decoders.RegisterF(int32(cmd_pb.CommandType_SyncSnapshot), network.decodeSyncSnapshotCommand)

	routerCommPool, err := network.createRouterCommPool(config)
	if err != nil {
		return nil, err
	}
	network.Inspections = NewInspectionsManager(network)

	network.routerConnectSem = concurrenz.NewSemaphore(int(config.GetOptions().RouterConnectConcurrency))
	network.routerConnectWaitTimer = config.GetMetricsRegistry().Timer(RouterConnectWaitTimer)
	network.routerConnectWorkTimer = config.GetMetricsRegistry().Timer(RouterConnectWorkTimer)
	network.routerConnectTimeouts = config.GetMetricsRegistry().Meter(RouterConnectTimeoutMeter)
	// A func gauge rather than a counter the code updates: the depth is derived from a value already
	// maintained, so there is nothing to keep in step.
	config.GetMetricsRegistry().FuncGauge(RouterConnectQueueSizeGauge, func() int64 {
		return network.routerConnectWaiting.Load()
	})
	network.RouterMessaging = NewRouterMessaging(env, routerCommPool)

	env.GetManagers().Router.Store.AddEntityIdListener(network.HandleRouterDelete, boltz.EntityDeletedAsync)

	network.AddCapability("ziti.fabric")
	network.showOptions()
	network.relayControllerMetrics()
	servermetrics.RegisterHostStats(network.metricsRegistry, servermetrics.HostStatsConfig{
		Enabled: config.GetOptions().HostMetrics.Enabled,
	})

	// ctrl.is_leader is 1 on the raft leader, 0 otherwise (always 1 in non-HA).
	// Lets the metrics timeseries correlate per-controller load with leadership,
	// which matters under chaos/partitions that shuffle leadership. The closure
	// reads IsLeader() dynamically, so it tracks leadership set later by HA init.
	network.metricsRegistry.FuncGauge("ctrl.is_leader", func() int64 {
		if network.IsLeader() {
			return 1
		}
		return 0
	})
	network.AddRouterPresenceHandler(network.RouterMessaging)
	go network.RouterMessaging.run()

	return network, nil
}

// IsLeader reports whether this controller speaks for the cluster: always true where raft is not
// enabled, the raft leader where it is. For picking a single controller to do work that must not be
// done by several, and for reporting which one is currently in that position.
func (network *Network) IsLeader() bool {
	if !network.config.IsRaftEnabled() {
		return true
	}
	return network.config.IsRaftLeader()
}

func (self *Network) HandleRouterDelete(id string) {
	self.routerDeleted(id)
	self.RouterMessaging.RouterDeleted(id)
	// Before the index drop: retirement reads the per-router index to find what to retire, and dropping it
	// first leaves the links in place on the surviving endpoint.
	self.retireLinksForDeletedRouter(id)
	self.Link.RouterDeleted(id)
}

// retireLinksForDeletedRouter removes every link with the deleted router at either end.
//
// Dropping the router's index alone left those links in the link table and in the surviving endpoint's set,
// where nothing would remove them: a fleet that churns routers accumulated links to routers that were not
// there, and each one still counted, still validated and still offered itself for routing.
//
// Retired the way a fault is rather than merely deleted, so circuits using the link are rerouted and the
// removal reaches the event stream.
//
// Retires by id, and deliberately does not check whether the id has since come back. Store commit handlers
// run after bolt releases the writer lock, so this can land late; an id that was deleted and recreated in
// that window would have the new router's links retired. Reaching it needs a router created with a chosen
// id, which only the deprecated create-outside-enrollment path allows, and needs the delete and recreate to
// land inside that window. Guarding it is not worth the machinery.
func (network *Network) retireLinksForDeletedRouter(routerId string) {
	links := network.Link.LinksForRouter(routerId)
	if len(links) == 0 {
		return
	}

	for _, link := range links {
		network.LinkFaulted(link, false)
	}

	pfxlog.Logger().WithField("routerId", routerId).WithField("links", len(links)).
		Info("removed links belonging to a deleted router")
}

func (self *Network) decodeSyncSnapshotCommand(_ int32, data []byte) (command.Command, error) {
	msg := &cmd_pb.SyncSnapshotCommand{}
	if err := proto.Unmarshal(data, msg); err != nil {
		return nil, err
	}

	cmd := &command.SyncSnapshotCommand{
		TimelineId:   msg.SnapshotId,
		Snapshot:     msg.Snapshot,
		ClusterId:    msg.ClusterId,
		SnapshotSink: self.RestoreSnapshot,
	}

	return cmd, nil
}

func routerCommunicationsWorker(_ uint32, f func()) {
	f()
}

func (network *Network) createRouterCommPool(config Config) (goroutines.Pool, error) {
	poolConfig := goroutines.PoolConfig{
		QueueSize:   config.GetOptions().RouterComm.QueueSize,
		MinWorkers:  0,
		MaxWorkers:  config.GetOptions().RouterComm.MaxWorkers,
		IdleTime:    30 * time.Second,
		CloseNotify: config.GetCloseNotify(),
		PanicHandler: func(err interface{}) {
			pfxlog.Logger().WithField(logrus.ErrorKey, err).WithField("backtrace", string(debug.Stack())).Error("panic during message send to router")
		},
		WorkerFunction: routerCommunicationsWorker,
	}

	servermetrics.ConfigureGoroutinesPoolMetrics(&poolConfig, config.GetMetricsRegistry(), "pool.router.messaging")

	pool, err := goroutines.NewPool(poolConfig)
	if err != nil {
		return nil, fmt.Errorf("error creating router messaging pool (%w)", err)
	}
	return pool, nil
}

func (network *Network) relayControllerMetrics() {
	go func() {
		timer := time.NewTicker(network.options.MetricsReportInterval)
		defer timer.Stop()

		for {
			select {
			case <-timer.C:
				if msg := servermetrics.Poll(network.metricsRegistry); msg != nil {
					network.eventDispatcher.AcceptMetricsMsg(msg)
				}
			case <-network.closeNotify:
				return
			}
		}
	}()
}

func (network *Network) InitServiceCounterDispatch(handler servermetrics.Handler) {
	network.serviceEventMetrics.StartReporting(handler, network.GetOptions().MetricsReportInterval, 10)
}

func (network *Network) GetAppId() string {
	return network.nodeId
}

func (network *Network) GetOptions() *config.NetworkConfig {
	return network.options
}

func (network *Network) GetDb() boltz.Db {
	return network.config.GetDb()
}

func (network *Network) GetStores() *db.Stores {
	return network.env.GetStores()
}

func (network *Network) GetConnectedRouter(routerId string) *model.Router {
	return network.Router.GetConnected(routerId)
}

// NewCtrlChanRouter returns the Router instance representing a new control-channel connection, with the
// channel recorded on it. Each connection gets its own instance, which is what lets the connect and
// disconnect paths tell two racing connections for one router apart.
func (network *Network) NewCtrlChanRouter(ch channel.Channel) (*model.Router, error) {
	return network.Router.NewCtrlChanRouter(ch)
}

func (network *Network) GetRouter(routerId string) (*model.Router, error) {
	return network.Router.Read(routerId)
}

func (network *Network) AllConnectedRouters() []*model.Router {
	return network.Router.AllConnected()
}

func (network *Network) GetLink(linkId string) (*model.Link, bool) {
	return network.Link.Get(linkId)
}

func (network *Network) GetAllLinks() []*model.Link {
	return network.Link.All()
}

func (network *Network) GetAllLinksForRouter(routerId string) []*model.Link {
	r := network.GetConnectedRouter(routerId)
	if r == nil {
		return nil
	}
	return r.GetLinks()
}

func (network *Network) GetCircuit(circuitId string) (*model.Circuit, bool) {
	return network.Circuit.Get(circuitId)
}

func (network *Network) GetAllCircuits() []*model.Circuit {
	return network.Circuit.All()
}

func (network *Network) GetCircuitStore() *objectz.ObjectStore[*model.Circuit] {
	return network.Circuit.GetStore()
}

func (network *Network) GetLinkStore() *objectz.ObjectStore[*model.Link] {
	return network.Link.GetStore()
}

func (network *Network) RouteResult(rs *RouteStatus) bool {
	return network.routeSenderController.forwardRouteResult(rs)
}

func (network *Network) newRouteSender(circuitId string) *routeSender {
	rs := newRouteSender(circuitId, network.options.RouteTimeout, network, network.Terminator)
	network.routeSenderController.addRouteSender(rs)
	return rs
}

func (network *Network) removeRouteSender(rs *routeSender) {
	network.routeSenderController.removeRouteSender(rs)
}

func (network *Network) GetEventDispatcher() event.Dispatcher {
	return network.eventDispatcher
}

func (network *Network) GetTraceController() trace.Controller {
	return network.traceController
}

func (network *Network) GetMetricsRegistry() metrics.Registry {
	return network.metricsRegistry
}

func (network *Network) GetServiceEventsMetricsRegistry() servermetrics.UsageRegistry {
	return network.serviceEventMetrics
}

func (network *Network) GetCloseNotify() <-chan struct{} {
	return network.closeNotify
}

func (network *Network) ConnectedRouter(id string) bool {
	return network.Router.IsConnected(id)
}

var (
	// ErrConnectRejected indicates a router connect was rejected because another connection for the same
	// router is already current. It is returned by ConnectRouter and propagated out of the bind handler so
	// NewChannel closes the rejected connection's underlay without starting rx or registering it; the
	// router then redials.
	ErrConnectRejected = errors.New("router connect rejected: another connection is already current")

	// ErrConnectChannelClosed indicates a router connect was refused because its control channel was
	// already closed by the time the connect decision was made, so the connection must not be registered.
	ErrConnectChannelClosed = errors.New("router connect rejected: control channel already closed")
)

// IsConnectRejected reports whether err is (or wraps) a connect refusal that the router recovers from by
// redialing, so the accept path can log it at info rather than treating it as a bind failure.
func IsConnectRejected(err error) bool {
	return errors.Is(err, ErrConnectRejected) || errors.Is(err, ErrConnectChannelClosed)
}

// ConnectRouter registers r as the current connection for its router id, serialized per router. If the
// slot is already held by a different connection it rejects this one (returning ErrConnectRejected) and
// displaces the occupant so its teardown runs; the router redials into the freed slot. A connection whose
// channel is already closed is refused outright (ErrConnectChannelClosed) rather than registered. There is
// at most one connection per router in the connected map at a time.
// Router connect setup metrics. Named to match the goroutine pool convention the other gossip pools use
// (`pool.gossip.apply.*`, `pool.peer.events.*`, `pool.io.*`), so a dashboard built for one transfers, even
// though this path is bounded by a semaphore rather than a pool.
const (
	// RouterConnectQueueSizeGauge is the number of connects waiting for a slot. The one signal that shows
	// a wedge: the timers below only record on completion.
	RouterConnectQueueSizeGauge = "pool.router.connect.queue_size"

	// RouterConnectWaitTimer is how long a connect waited for a slot, which measures contention.
	RouterConnectWaitTimer = "pool.router.connect.wait_timer"

	// RouterConnectWorkTimer is how long the setup itself took, which measures the work. Split from the
	// wait because the two have opposite remedies: contention wants more concurrency, slow setup wants a
	// look at what the setup does.
	RouterConnectWorkTimer = "pool.router.connect.work_timer"

	// RouterConnectTimeoutMeter counts connects that gave up waiting and were closed to force a redial.
	RouterConnectTimeoutMeter = "pool.router.connect.timeouts"
)

// runConnectSetup waits for a slot and runs the connect setup, giving up if the wait exceeds the
// configured timeout.
//
// Giving up closes the router's control channel rather than skipping the setup. Skipping would leave a
// router that is connected, believes it is, and has no links built, which is invisible everywhere; closing
// makes it redial with its normal backoff and says so in the log. The timeout is long enough that reaching
// it means setups are wedged rather than slow, and waiters behind a wedge were never going to drain.
//
// Waiters are released oldest first, since the semaphore is a channel receive, so a timeout here is a
// statement about the whole queue rather than about one unlucky router.
func (network *Network) runConnectSetup(r *model.Router) {
	network.routerConnectWaiting.Add(1)
	waitStart := time.Now()
	acquired := network.routerConnectSem.AcquireWithTimeout(network.options.RouterConnectSetupTimeout)
	network.routerConnectWaiting.Add(-1)
	network.routerConnectWaitTimer.UpdateSince(waitStart)

	if !acquired {
		network.routerConnectTimeouts.Mark(1)
		log := pfxlog.Logger().
			WithField("routerId", r.Id).
			WithField("waitedMs", time.Since(waitStart).Milliseconds())
		if ctrl := r.Control; ctrl != nil {
			log.Error("timed out waiting to run router connect setup, closing the connection so the router redials")
			_ = ctrl.Close()
		} else {
			log.Error("timed out waiting to run router connect setup, and the router has no channel to close")
		}
		return
	}
	defer network.routerConnectSem.Release()

	setupStart := time.Now()
	network.ConnectRouter(r)
	network.routerConnectWorkTimer.UpdateSince(setupStart)
}

// QueueRouterConnect makes the synchronous connect decision for r, serialized per router. If the slot is
// already held by a different connection it rejects this one (returning ErrConnectRejected) and displaces
// the occupant so its teardown runs; the router redials into the freed slot. A connection whose channel is
// already closed is refused outright (ErrConnectChannelClosed) rather than registered. Otherwise it
// registers r (MarkConnected), fires synchronous presence handlers, and hands the remaining, non-gating
// setup to a goroutine bounded by the connect semaphore. The decision and registration are synchronous so
// the bind handler can fail a refused connect (no rx started, not registered) and so there is at most one
// connection per router current at a time. Capacity is never a reason to refuse: a refusal fails the bind,
// which strands the router's channel group, so the setup waits for a slot instead.
func (network *Network) QueueRouterConnect(r *model.Router) error {
	unlock := network.Router.LockConnectFor(r.Id)
	defer unlock() // leak-safety net; idempotent, so the explicit unlock below is the one that matters

	if cur := network.Router.GetConnected(r.Id); cur != nil && cur != r {
		// Displace the occupant outside the lock: the teardown acquires the stripe itself (we have
		// released it), so there is no reentrant self-deadlock.
		unlock()
		network.displaceConnection(cur)
		return ErrConnectRejected
	}

	// Its close handler has already run and never fires again, so nothing would remove it from the
	// connected map and every redial would bounce off a slot that can never be freed.
	if r.Control == nil || r.Control.IsClosed() {
		return ErrConnectChannelClosed
	}

	// Mark connected before the deferred setup builds this router's links: repairDest re-reads the
	// connected map for links that land in the table after that build, and must find the router there.
	network.Router.MarkConnected(r)

	for _, h := range network.routerPresenceHandlers.Value() {
		if syncCapableHandler, ok := h.(model.SyncRouterPresenceHandler); ok && syncCapableHandler.InvokeRouterConnectedSynchronously() {
			h.RouterConnected(r)
		}
	}
	unlock()

	// Off the accept goroutine, so binding stays quick: the listener holds this group's create
	// reservation until the bind returns, and the group's other underlays wait on it. The semaphore
	// bounds how many of these run at once so a mass reconnect cannot starve the gossip and I/O pools,
	// and it waits rather than refusing, since a refusal here would fail the bind, and the expensive
	// part of connecting has already been paid by this point.
	go network.runConnectSetup(r)

	return nil
}

// ConnectRouter runs the deferred, non-gating connect setup for r on the router connect pool. It
// re-acquires the per-router lock and re-checks currency, so its link building and gossip reconcile are
// serialized against a concurrent disconnect of the same router and are skipped entirely if r was
// superseded or disconnected while it was queued. Different routers run concurrently on separate stripes.
func (network *Network) ConnectRouter(r *model.Router) {
	unlock := network.Router.LockConnectFor(r.Id)
	if !network.isCurrentConnection(r) {
		unlock()
		return
	}

	network.Link.BuildRouterLinks(r)

	for _, h := range network.routerPresenceHandlers.Value() {
		if syncHandler, ok := h.(model.SyncRouterPresenceHandler); ok && syncHandler.InvokeRouterConnectedSynchronously() {
			continue // already called synchronously in QueueRouterConnect
		}
		h.RouterConnected(r)
	}
	unlock()

	network.ValidateTerminators(r)
}

func (network *Network) ValidateTerminators(r *model.Router) {
	logger := pfxlog.Logger().WithField("routerId", r.Id)
	result, err := network.Terminator.Query(fmt.Sprintf(`router.id = "%v" limit none`, r.Id))
	if err != nil {
		logger.WithError(err).Error("failed to get terminators for router")
		return
	}

	logger.Debugf("%v terminators to validate", len(result.Entities))
	if len(result.Entities) == 0 {
		return
	}

	network.RouterMessaging.ValidateRouterTerminators(result.Entities)
}

type LinkValidationCallback func(detail *mgmt_pb.RouterLinkDetails)

func (n *Network) ValidateLinks(filter string, cb LinkValidationCallback) (int64, func(), error) {
	result, err := n.Router.BaseList(filter)
	if err != nil {
		return 0, nil, err
	}

	sem := concurrenz.NewSemaphore(10)

	evalF := func() {
		for _, router := range result.Entities {
			connectedRouter := n.GetConnectedRouter(router.Id)
			if connectedRouter != nil {
				sem.Acquire()
				go func() {
					defer sem.Release()
					n.ValidateRouterLinks(connectedRouter, cb)
				}()
			} else {
				n.reportRouterLinksError(router, errors.New("router not connected"), cb)
			}
		}
	}

	return int64(len(result.Entities)), evalF, nil
}

type SdkTerminatorValidationCallback func(detail *mgmt_pb.RouterSdkTerminatorsDetails)

func (n *Network) ValidateRouterSdkTerminators(filter string, cb SdkTerminatorValidationCallback) (int64, func(), error) {
	result, err := n.Router.BaseList(filter)
	if err != nil {
		return 0, nil, err
	}

	sem := concurrenz.NewSemaphore(10)

	evalF := func() {
		for _, router := range result.Entities {
			connectedRouter := n.GetConnectedRouter(router.Id)
			if connectedRouter != nil {
				sem.Acquire()
				go func() {
					defer sem.Release()
					n.Router.ValidateRouterSdkTerminators(connectedRouter, cb)
				}()
			} else {
				n.Router.ReportRouterSdkTerminatorsError(router, errors.New("router not connected"), cb)
			}
		}
	}

	return int64(len(result.Entities)), evalF, nil
}

type ErtTerminatorValidationCallback func(detail *mgmt_pb.RouterErtTerminatorsDetails)

func (n *Network) ValidateRouterErtTerminators(filter string, cb ErtTerminatorValidationCallback) (int64, func(), error) {
	result, err := n.Router.BaseList(filter)
	if err != nil {
		return 0, nil, err
	}

	sem := concurrenz.NewSemaphore(10)

	evalF := func() {
		for _, router := range result.Entities {
			connectedRouter := n.GetConnectedRouter(router.Id)
			if connectedRouter != nil {
				sem.Acquire()
				go func() {
					defer sem.Release()
					n.Router.ValidateRouterErtTerminators(connectedRouter, cb)
				}()
			} else {
				n.Router.ReportRouterErtTerminatorsError(router, errors.New("router not connected"), cb)
			}
		}
	}

	return int64(len(result.Entities)), evalF, nil
}

// isCurrentConnection reports whether r is still the router's current, connected connection, by pointer
// identity against the connected map (mirrors the check in NotifyExistingLink). A stale or superseded
// connection returns false.
func (network *Network) isCurrentConnection(r *model.Router) bool {
	return network.Router.GetConnected(r.Id) == r && r.Connected.Load()
}

// displaceConnection removes cur, the connection occupying its router's connected slot, so that a redial
// can take the slot. Closing the channel is not sufficient on its own: if it is already closed, its close
// handler has already run and will never run again, so nothing would remove cur and every subsequent
// connect would be rejected against a slot that can never be freed. The teardown is therefore also
// invoked directly; it is gated on connection currency, so it is a no-op once the close handler has
// cleared the slot. Must be called with the router's connect stripe released, since the teardown
// acquires it.
func (network *Network) displaceConnection(cur *model.Router) {
	if ch := cur.Control; ch != nil && !ch.IsClosed() {
		if err := ch.Close(); err != nil {
			pfxlog.Logger().WithError(err).WithField("routerId", cur.Id).
				Error("error closing superseded control channel while rejecting connect")
		}
	}
	network.DisconnectRouter(cur)
}

func (network *Network) DisconnectRouter(r *model.Router) {
	// Lock-free pre-check: a stale/superseded disconnect (e.g. the old connection after a takeover) has
	// nothing to tear down and must not touch the live connection's state; bail without blocking.
	if !network.isCurrentConnection(r) {
		return
	}

	unlock := network.Router.LockConnectFor(r.Id)
	defer unlock()

	// Re-check under the stripe: a newer connection may have taken over between the pre-check and the
	// lock. The teardown is all-or-nothing and must not run against a superseded connection.
	if !network.isCurrentConnection(r) {
		return
	}

	// Snapshot the router's links before marking it disconnected: MarkDisconnected clears the
	// router's link set (routerLinks.Clear()), so a later r.GetLinks() would return nothing and
	// the link-removal/reroute cascade below would be skipped entirely.
	links := r.GetLinks()

	// Mark the router disconnected before the RerouteLink cascade, so reroute and everything it
	// calls (shortestPath, connected-map reads) sees the dying router as gone. Otherwise reroute
	// runs while the router still appears connected and can compute a replacement path through the
	// very router that is being removed.
	network.Router.MarkDisconnected(r)

	// A router that marks its reconnect announcement a full refresh says, authoritatively, which links it
	// still has, and the controller prunes the rest against it. There is nothing left here for a disconnect
	// to clean up, and retiring anyway is worse than useless: the announcement is a bounded, best-effort
	// send, so an outage that outlives its retries leaves this controller with no links for a router that
	// is up, until it happens to reconnect again.
	retireUnannounced := !r.SendsFullRefresh()

	// remove Links for Router, rerouting circuits off any that were connected
	for _, l := range links {
		wasConnected := l.CurrentState().Mode == model.Connected
		if l.Src.Id != r.Id {
			continue
		}

		// A link its dialer does not prune itself is known only from that router's reports, under ids it
		// will never mention again. This is the only removal such a link will get.
		if retireUnannounced {
			network.LinkFaulted(l, false)
			continue
		}
		if wasConnected {
			network.RerouteLink(l)
		}
	}

	for _, h := range network.routerPresenceHandlers.Value() {
		h.RouterDisconnected(r)
	}
}

func (network *Network) NotifyExistingLink(srcRouter *model.Router, reportedLink *ctrl_pb.RouterLinks_RouterLink) {
	log := pfxlog.Logger().
		WithField("routerId", srcRouter.Id).
		WithField("linkId", reportedLink.Id).
		WithField("destRouterId", reportedLink.DestRouterId).
		WithField("iteration", reportedLink.Iteration)

	// Publish under the stripe DisconnectRouter holds: checking currency and then publishing without it
	// lets a report recreate a link after the teardown has snapshotted and cleared it. Events go out after
	// the unlock, since a dispatcher may be slow and this stripe is shared with connect and disconnect.
	unlock := network.Router.LockConnectFor(srcRouter.Id)

	src := network.Router.GetConnected(srcRouter.Id)
	if src == nil {
		unlock()
		log.Info("ignoring links message processed after router disconnected")
		return
	}

	if src != srcRouter || !srcRouter.Connected.Load() {
		unlock()
		log.Info("ignoring links message processed from old router connection")
		return
	}

	dst := network.Router.GetConnected(reportedLink.DestRouterId)
	link, created := network.Link.RouterReportedLink(reportedLink, src, dst)

	unlock()

	if dst == nil {
		network.NotifyLinkIdEvent(reportedLink.Id, event.LinkFromRouterDisconnectedDest)
	}

	if link == nil {
		// Refused; RouterReportedLink has already said why. A report naming a destination that no longer
		// exists is also the evidence that the router never got the removal, so answer it with one.
		network.notifyDestinationRemovedIfGone(src, reportedLink.DestRouterId, dst)
		return
	}

	if created {
		network.NotifyLinkEvent(link, event.LinkFromRouterNew)
		log.Info("router reported link added")
	} else {
		network.NotifyLinkEvent(link, event.LinkFromRouterKnown)
		log.Info("router reported link already known")
	}
}

// notifyDestinationRemovedIfGone tells srcRouter that destRouterId is gone, when a link report or gossip
// entry naming it has been refused. dst is the destination's connection here, if any.
func (network *Network) notifyDestinationRemovedIfGone(srcRouter *model.Router, destRouterId string, dst *model.Router) {
	if network.shouldNotifyDestinationRemoved(srcRouter, destRouterId, dst) {
		network.RouterMessaging.NotifyRouterRemoved(srcRouter.Id, destRouterId)
	}
}

// shouldNotifyDestinationRemoved reports whether to re-assert a removal for destRouterId to srcRouter. dst
// is the destination's connection here, if any.
//
// Peer state is only recomputed when a router connects or disconnects, and a deleted router does neither
// again, so a removal a router misses is never repeated and it dials the dead destination forever. A link
// report or a gossip entry naming the destination is the evidence that it was missed, and both repeat, so
// the correction repeats with them.
//
// Every router that names the destination is told, whatever it supports. A router in link gossip learns of
// deletions from the router data model instead, but only where its controllers distribute routers there,
// and the statement is true regardless of who hears it.
//
// Leader only. This is an unsolicited assertion rather than a reaction to a local event, and it tears links
// down: a controller lagging on a create would otherwise tell a router to drop a destination that is fine.
// One controller saying it is enough, since the reports that prompt it repeat.
func (network *Network) shouldNotifyDestinationRemoved(srcRouter *model.Router, destRouterId string, dst *model.Router) bool {
	if srcRouter == nil || destRouterId == "" || dst != nil {
		return false
	}

	if !network.IsLeader() {
		return false
	}

	// A gossip entry names its dialer whether or not that router is connected here, and a router this
	// controller does not hold is one it cannot tell.
	if network.Router.GetConnected(srcRouter.Id) == nil {
		return false
	}

	exists, err := network.Router.Exists(destRouterId)
	if err != nil {
		// Saying a router is gone on the strength of a failed read would have the reporting router tear
		// down links to one that is probably fine.
		pfxlog.Logger().WithError(err).WithField("routerId", destRouterId).
			Error("failed to check if reported link destination exists")
		return false
	}
	return !exists
}

func (network *Network) LinkFaulted(link *model.Link, dupe bool) {
	wasUsable := link.IsUsable()

	link.SetState(model.Failed)
	if dupe {
		network.NotifyLinkEvent(link, event.LinkDuplicate)
	} else {
		network.NotifyLinkEvent(link, event.LinkFault)
	}

	pfxlog.Logger().WithField("linkId", link.Id).Info("removing failed link")
	network.Link.Remove(link)

	if wasUsable {
		network.RerouteLink(link)
	}
}

func (network *Network) VerifyRouter(routerId string, fingerprints []string) error {
	router, err := network.GetRouter(routerId)
	if err != nil {
		return err
	}

	routerFingerprint := router.Fingerprint
	if routerFingerprint == nil {
		return fmt.Errorf("invalid router %v, not yet enrolled", routerId)
	}

	for _, fp := range fingerprints {
		if fp == *routerFingerprint {
			return nil
		}
	}

	return fmt.Errorf("could not verify fingerprint for router %v", routerId)
}

func (network *Network) RerouteLink(l *model.Link) {
	// This is called from Channel.rxer() and thus may not block
	go func() {
		network.handleRerouteLink(l)
	}()
}

func (network *Network) CreateCircuit(params model.CreateCircuitParams) (*model.Circuit, error) {
	clientId := params.GetClientId()
	service := params.GetServiceId()
	ctx := params.GetLogContext()
	deadline := params.GetDeadline()

	startTime := time.Now()

	instanceId, serviceId := parseInstanceIdAndService(service)

	// 1: Allocate Circuit Identifier
	circuitId := params.GetCircuitId()
	if circuitId == "" {
		var err error
		circuitId, err = idgen.NewUUIDString()
		if err != nil {
			network.CircuitFailedEvent(circuitId, params, startTime, nil, nil, CircuitFailureIdGenerationError)
			return nil, err
		}
	}
	ctx.WithFields(map[string]interface{}{
		"circuitId":     circuitId,
		"serviceId":     service,
		"attemptNumber": 1,
	})
	logger := pfxlog.ChannelLogger(logcontext.SelectPath).Wire(ctx).Entry

	circuit := &model.Circuit{
		Id:        circuitId,
		ClientId:  clientId.Token,
		ServiceId: serviceId,
		Path:      &model.Path{}, // empty until the circuit is built
	}

	// Reserve the circuit ID in the map to prevent collisions and protect routing.
	// On any error path, the deferred cleanup removes the stub.
	if !network.Circuit.Reserve(circuit) {
		return nil, fmt.Errorf("circuit id %v already in use", circuitId)
	}
	removeReserved := true
	defer func() {
		if removeReserved {
			network.Circuit.Remove(circuit)
		}
	}()

	attempt := uint32(0)
	allCleanups := make(map[string]struct{})
	rs := network.newRouteSender(circuitId)
	defer func() { network.removeRouteSender(rs) }()
	for {
		// 2: Find Service
		svc, err := network.Service.Read(serviceId)
		if err != nil {
			network.CircuitFailedEvent(circuitId, params, startTime, nil, nil, CircuitFailureInvalidService)
			network.ServiceDialOtherError(serviceId)
			return circuit, err
		}
		logger = logger.WithField("serviceName", svc.Name)

		// 3: select terminator
		strategy, terminator, pathNodes, strategyData, circuitErr := network.selectPath(params, svc, instanceId, ctx)
		if circuitErr != nil {
			network.CircuitFailedEvent(circuitId, params, startTime, nil, nil, circuitErr.Cause())
			network.ServiceDialOtherError(serviceId)
			return circuit, circuitErr
		}

		circuit.Terminator = terminator

		// 4: Create Path
		path, pathErr := network.CreatePathWithNodes(pathNodes)
		if pathErr != nil {
			network.CircuitFailedEvent(circuitId, params, startTime, nil, terminator, pathErr.Cause())
			network.ServiceDialOtherError(serviceId)
			return circuit, pathErr
		}

		circuit.Path = path

		// get circuit tags
		tags := params.GetCircuitTags(terminator)

		circuit.Tags = tags

		// 4a: Create Route Messages
		rms := network.CreateRouteMessages(path, attempt, circuitId, terminator, deadline)
		rms[len(rms)-1].Egress.PeerData = clientId.Data
		for _, msg := range rms {
			msg.Context = &ctrl_pb.Context{
				Fields:      ctx.GetStringFields(),
				ChannelMask: ctx.GetChannelsMask(),
			}
			msg.Tags = tags
		}

		// 5: Routing
		logger.Debug("route attempt for circuit")
		peerData, cleanups, circuitErr := rs.route(attempt, path, rms, strategy, terminator, ctx.Clone())
		for k, v := range cleanups {
			allCleanups[k] = v
		}
		if circuitErr != nil {
			logger.WithError(circuitErr).Warn("route attempt for circuit failed")
			network.CircuitFailedEvent(circuitId, params, startTime, path, terminator, circuitErr.Cause())
			attempt++
			ctx.WithField("attemptNumber", attempt)
			logger = logger.WithField("attemptNumber", attempt)
			if attempt < network.options.CreateCircuitRetries {
				continue
			} else {
				// revert successful routes
				logger.Warnf("circuit creation failed after [%d] attempts, sending cleanup unroutes", network.options.CreateCircuitRetries)
				for cleanupRId := range allCleanups {
					if r := network.GetConnectedRouter(cleanupRId); r != nil {
						if err := sendUnroute(r, circuitId, true); err == nil {
							logger.WithField("routerId", cleanupRId).Debug("sent cleanup unroute for circuit")
						} else {
							logger.WithField("routerId", cleanupRId).Error("error sending cleanup unroute for circuit")
						}
					} else {
						logger.WithField("routerId", cleanupRId).Error("router for circuit cleanup not connected")
					}
				}

				return circuit, fmt.Errorf("exceeded maximum [%d] retries creating circuit [c/%s] (%w)", network.options.CreateCircuitRetries, circuitId, circuitErr)
			}
		}

		// 5.a: Unroute Abandoned Routers (from Previous Attempts)
		usedRouters := make(map[string]struct{})
		for _, r := range path.Nodes {
			usedRouters[r.Id] = struct{}{}
		}
		cleanupCount := 0
		for cleanupRId := range allCleanups {
			if _, found := usedRouters[cleanupRId]; !found {
				cleanupCount++
				if r := network.GetConnectedRouter(cleanupRId); r != nil {
					if err := sendUnroute(r, circuitId, true); err == nil {
						logger.WithField("routerId", cleanupRId).Debug("sent abandoned cleanup unroute for circuit to router")
					} else {
						logger.WithField("routerId", cleanupRId).WithError(err).Error("error sending abandoned cleanup unroute for circuit to router")
					}
				} else {
					logger.WithField("routerId", cleanupRId).Error("router not connected for circuit, abandoned cleanup")
				}
			}
		}
		logger.Debugf("cleaned up [%d] abandoned routers for circuit", cleanupCount)

		path.InitiatorLocalAddr = string(clientId.Data[uint32(ctrl_msg.InitiatorLocalAddressHeader)])
		path.InitiatorRemoteAddr = string(clientId.Data[uint32(ctrl_msg.InitiatorRemoteAddressHeader)])
		path.TerminatorLocalAddr = string(peerData[uint32(ctrl_msg.TerminatorLocalAddressHeader)])
		path.TerminatorRemoteAddr = string(peerData[uint32(ctrl_msg.TerminatorRemoteAddressHeader)])

		delete(peerData, uint32(ctrl_msg.InitiatorLocalAddressHeader))
		delete(peerData, uint32(ctrl_msg.InitiatorRemoteAddressHeader))
		delete(peerData, uint32(ctrl_msg.TerminatorLocalAddressHeader))
		delete(peerData, uint32(ctrl_msg.TerminatorRemoteAddressHeader))

		for k, v := range strategyData {
			peerData[k] = v
		}

		now := time.Now()
		// 6: Create Circuit Object
		circuit.PeerData = peerData
		circuit.CreatedAt = now
		circuit.UpdatedAt = now

		removeReserved = false // circuit is finalized, don't remove on defer
		creationTimespan := time.Since(startTime)
		network.CircuitEvent(event.CircuitCreated, circuit, &creationTimespan)
		strategy.NotifyEvent(xt.NewDialSucceeded(terminator))

		logger.WithField("path", circuit.Path).
			WithField("terminator_local_address", circuit.Path.TerminatorLocalAddr).
			WithField("terminator_remote_address", circuit.Path.TerminatorRemoteAddr).
			Debug("created circuit")
		return circuit, nil
	}
}

func (network *Network) ReportForwardingFaults(ffr *ForwardingFaultReport) {
	go network.handleForwardingFaults(ffr)
}

func parseInstanceIdAndService(service string) (string, string) {
	atIndex := strings.IndexRune(service, '@')
	if atIndex < 0 {
		return "", service
	}
	identityId := service[0:atIndex]
	serviceId := service[atIndex+1:]
	return identityId, serviceId
}

func (network *Network) selectPath(params model.CreateCircuitParams, svc *model.Service, instanceId string, ctx logcontext.Context) (xt.Strategy, xt.CostedTerminator, []*model.Router, xt.PeerData, CircuitError) {
	paths := map[string]*PathAndCost{}
	var weightedTerminators []xt.CostedTerminator
	var errList []error

	log := pfxlog.ChannelLogger(logcontext.SelectPath).Wire(ctx)

	hasOfflineRouters := false
	pathError := false

	for _, terminator := range svc.Terminators {
		if terminator.InstanceId != instanceId {
			continue
		}

		pathAndCost, found := paths[terminator.Router]
		if !found {
			dstR := network.Router.GetConnected(terminator.GetRouterId())
			if dstR == nil {
				err := fmt.Errorf("router with id=%v on terminator with id=%v for service name=%v is not online",
					terminator.GetRouterId(), terminator.GetId(), svc.Name)
				log.Debugf("error while calculating path for service %v: %v", svc.Id, err)

				errList = append(errList, err)
				hasOfflineRouters = true
				continue
			}

			path, cost, err := network.shortestPath(params.GetSourceRouter(), dstR)
			if err != nil {
				log.Debugf("error while calculating path for service %v: %v", svc.Id, err)
				errList = append(errList, err)
				pathError = true
				continue
			}

			pathAndCost = newPathAndCost(path, cost)
			paths[terminator.GetRouterId()] = pathAndCost
		}

		dynamicCost := xt.GlobalCosts().GetDynamicCost(terminator.Id)
		unbiasedCost := uint32(terminator.Cost) + uint32(dynamicCost) + pathAndCost.cost
		biasedCost := terminator.Precedence.GetBiasedCost(unbiasedCost)
		costedTerminator := &model.RoutingTerminator{
			Terminator: terminator,
			RouteCost:  biasedCost,
		}
		weightedTerminators = append(weightedTerminators, costedTerminator)
	}

	if len(svc.Terminators) == 0 {
		return nil, nil, nil, nil, newCircuitErrorf(CircuitFailureNoTerminators, "service %v has no terminators", svc.Id)
	}

	if len(weightedTerminators) == 0 {
		if pathError {
			return nil, nil, nil, nil, newCircuitErrWrap(CircuitFailureNoPath, errors.Join(errList...))
		}

		if hasOfflineRouters {
			return nil, nil, nil, nil, newCircuitErrorf(CircuitFailureNoOnlineTerminators, "service %v has no online terminators for instanceId %v", svc.Id, instanceId)
		}

		return nil, nil, nil, nil, newCircuitErrorf(CircuitFailureNoTerminators, "service %v has no terminators for instanceId %v", svc.Id, instanceId)
	}

	strategy, err := network.strategyRegistry.GetStrategy(svc.TerminatorStrategy)
	if err != nil {
		return nil, nil, nil, nil, newCircuitErrWrap(CircuitFailureInvalidStrategy, err)
	}

	sort.Slice(weightedTerminators, func(i, j int) bool {
		return weightedTerminators[i].GetRouteCost() < weightedTerminators[j].GetRouteCost()
	})

	terminator, peerData, err := strategy.Select(params, weightedTerminators)

	if err != nil {
		return nil, nil, nil, nil, newCircuitErrorf(CircuitFailureStrategyError, "strategy %v errored selecting terminator for service %v: %v", svc.TerminatorStrategy, svc.Id, err)
	}

	if terminator == nil {
		return nil, nil, nil, nil, newCircuitErrorf(CircuitFailureStrategyError, "strategy %v did not select terminator for service %v", svc.TerminatorStrategy, svc.Id)
	}

	path := paths[terminator.GetRouterId()].path

	if log.Logger.IsLevelEnabled(logrus.DebugLevel) {
		buf := strings.Builder{}
		buf.WriteString("[")
		if len(weightedTerminators) > 0 {
			fmt.Fprintf(&buf, "%v: %v", weightedTerminators[0].GetId(), weightedTerminators[0].GetRouteCost())
			for _, t := range weightedTerminators[1:] {
				buf.WriteString(", ")
				fmt.Fprintf(&buf, "%v: %v", t.GetId(), t.GetRouteCost())
			}
		}
		buf.WriteString("]")
		var routerIds []string
		for _, r := range path {
			routerIds = append(routerIds, fmt.Sprintf("r/%s", r.Id))
		}
		pathStr := strings.Join(routerIds, "->")
		log.Debugf("selected terminator %v for path %v from %v", terminator.GetId(), pathStr, buf.String())
	}

	return strategy, terminator, path, peerData, nil
}

func (network *Network) RemoveCircuit(circuitId string, now bool) error {
	log := pfxlog.Logger().WithField("circuitId", circuitId)

	if circuit, found := network.Circuit.Get(circuitId); found {
		for _, r := range circuit.Path.Nodes {
			err := sendUnroute(r, circuit.Id, now)
			if err != nil {
				log.Errorf("error sending unroute to [r/%s] (%s)", r.Id, err)
			}
		}

		network.Circuit.Remove(circuit)
		network.CircuitEvent(event.CircuitDeleted, circuit, nil)

		if svc, err := network.Service.Read(circuit.ServiceId); err == nil {
			if strategy, err := network.strategyRegistry.GetStrategy(svc.TerminatorStrategy); strategy != nil {
				strategy.NotifyEvent(xt.NewCircuitRemoved(circuit.Terminator))
			} else if err != nil {
				log.WithError(err).WithField("terminatorStrategy", svc.TerminatorStrategy).Warn("failed to notify strategy of circuit end, invalid strategy")
			}
		} else {
			log.WithError(err).Error("unable to get service for circuit")
		}

		log.Debug("removed circuit")

		return nil
	}
	return InvalidCircuitError{circuitId: circuitId}
}

func (network *Network) CreatePath(srcR, dstR *model.Router) (*model.Path, error) {
	ingressId, err := idgen.NewUUIDString()
	if err != nil {
		return nil, err
	}

	egressId, err := idgen.NewUUIDString()
	if err != nil {
		return nil, err
	}

	path := &model.Path{
		Links:     make([]*model.Link, 0),
		IngressId: ingressId,
		EgressId:  egressId,
		Nodes:     make([]*model.Router, 0),
	}
	path.Nodes = append(path.Nodes, srcR)
	path.Nodes = append(path.Nodes, dstR)

	return network.UpdatePath(path)
}

func (network *Network) setLinks(path *model.Path) error {
	if len(path.Nodes) > 1 {
		for i := 0; i < len(path.Nodes)-1; i++ {
			if link, found := network.Link.LeastExpensiveLink(path.Nodes[i], path.Nodes[i+1]); found {
				path.Links = append(path.Links, link)
			} else {
				return fmt.Errorf("no link from r/%v to r/%v", path.Nodes[i].Id, path.Nodes[i+1].Id)
			}
		}
	}
	return nil
}

func (network *Network) AddRouterPresenceHandler(h model.RouterPresenceHandler) {
	network.routerPresenceHandlers.Append(h)
}

func (network *Network) Run() {
	defer logrus.Info("exited")
	logrus.Info("started")

	go network.watchdog()

	ticker := time.NewTicker(time.Duration(network.options.CycleSeconds) * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			network.clean()
			network.smart()
			network.Link.ScanForDeadLinks(nil)

		case <-network.closeNotify:
			network.eventDispatcher.RemoveMetricsMessageHandler(network)
			network.metricsRegistry.DisposeAll()
			return
		}

		// notify the watchdog that we're processing
		select {
		case network.watchdogCh <- struct{}{}:
		default:
		}
	}
}

func (network *Network) watchdog() {
	watchdogInterval := 2 * time.Duration(network.options.CycleSeconds) * time.Second
	consecutiveFails := 0

	watchDogTicker := time.NewTicker(watchdogInterval)
	defer watchDogTicker.Stop()

	for {
		// check every 2x cycle seconds
		select {
		case <-watchDogTicker.C:
		case <-network.closeNotify:
			return
		}

		select {
		case <-network.watchdogCh:
			consecutiveFails = 0
			continue
		case <-network.closeNotify:
			return
		default:
			consecutiveFails++
			// network.Run didn't complete, something is stalling it
			pfxlog.Logger().
				WithField("watchdogInterval", watchdogInterval.String()).
				WithField("consecutiveFails", consecutiveFails).
				Warn("network.Run did not finish within watchdog interval")

			if consecutiveFails == 3 {
				debugz.DumpStack()
			}
		}
	}
}

func (network *Network) handleRerouteLink(l *model.Link) {
	log := logrus.WithField("linkId", l.Id)
	log.Info("changed link")
	if err := network.rerouteLink(l, time.Now().Add(config.DefaultOptionsRouteTimeout)); err != nil {
		log.WithError(err).Error("unexpected error rerouting link")
	}
}

func (network *Network) handleForwardingFaults(ffr *ForwardingFaultReport) {
	network.fault(ffr)
}

func (network *Network) AddCapability(capability string) {
	network.lock.Lock()
	defer network.lock.Unlock()
	network.capabilities = append(network.capabilities, capability)
}

func (network *Network) GetCapabilities() []string {
	network.lock.Lock()
	defer network.lock.Unlock()
	return network.capabilities
}

func (network *Network) RemoveLink(linkId string) {
	log := pfxlog.Logger().WithField("linkId", linkId)

	link, _ := network.Link.Get(linkId)
	var iteration uint32

	var routerList []*model.Router
	if link != nil {
		iteration = link.Iteration
		routerList = []*model.Router{link.Src}
		if dst := link.GetDest(); dst != nil {
			routerList = append(routerList, dst)
		}
		log = log.WithField("srcRouterId", link.Src.Id).
			WithField("dstRouterId", link.DstId).
			WithField("iteration", iteration)
		log.Info("deleting known link")
	} else {
		routerList = network.AllConnectedRouters()
		log.Info("deleting unknown link (sending link fault to all connected routers)")
	}

	for _, router := range routerList {
		fault := &ctrl_pb.Fault{
			Subject:   ctrl_pb.FaultSubject_LinkFault,
			Id:        linkId,
			Iteration: iteration,
		}

		if ctrl := router.Control; ctrl != nil {
			if err := protobufs.MarshalTyped(fault).WithTimeout(15 * time.Second).Send(ctrl.GetDefaultSender()); err != nil {
				log.WithField("faultDestRouterId", router.Id).WithError(err).
					Error("failed to send link fault to router on link removal")
			} else {
				log.WithField("faultDestRouterId", router.Id).WithError(err).
					Info("sent link fault to router on link removal")
			}
		}
	}

	if link != nil {
		network.Link.Remove(link)
		network.RerouteLink(link)
	}
}

// RemoveLinkAtIteration removes linkId only while it is still at iteration, and
// faults that iteration rather than whatever is current. Returns false, having
// done nothing, when the link is already gone or has been re-dialed since.
//
// For callers acting on a verdict that was computed earlier: a link id is reused
// across re-dials, so removing on a stale verdict would close the healthy
// replacement. The faulted iteration is the one that was actually judged, which
// also lets the receiving router reject it if the link moved on in the meantime.
func (network *Network) RemoveLinkAtIteration(linkId string, iteration uint32) bool {
	log := pfxlog.Logger().WithField("linkId", linkId).WithField("iteration", iteration)

	link, _ := network.Link.Get(linkId)
	if link == nil {
		log.Info("not removing link, no longer present")
		return false
	}
	if link.Iteration != iteration {
		log.WithField("currentIteration", link.Iteration).
			Info("not removing link, iteration has moved on since it was judged")
		return false
	}

	routerList := []*model.Router{link.Src}
	if dst := link.GetDest(); dst != nil {
		routerList = append(routerList, dst)
	}
	log = log.WithField("srcRouterId", link.Src.Id).WithField("dstRouterId", link.DstId)

	// Removal is the decision point, not the iteration check above: a newer
	// incarnation can be installed in between, and only the link table can
	// settle that atomically. Fault and reroute after it has, so a lost race
	// leaves the replacement untouched and reports no removal.
	if !network.Link.Remove(link) {
		log.Info("not removing link, iteration changed while it was being removed")
		return false
	}

	log.Info("deleted known link")
	network.sendLinkFaults(log, routerList, linkId, iteration)
	network.RerouteLink(link)
	return true
}

// sendLinkFaults tells each router in routerList that linkId at iteration has
// faulted. Best-effort: failures are logged, not returned.
func (network *Network) sendLinkFaults(log *logrus.Entry, routerList []*model.Router, linkId string, iteration uint32) {
	for _, router := range routerList {
		fault := &ctrl_pb.Fault{
			Subject:   ctrl_pb.FaultSubject_LinkFault,
			Id:        linkId,
			Iteration: iteration,
		}

		if ctrl := router.Control; ctrl != nil {
			if err := protobufs.MarshalTyped(fault).WithTimeout(15 * time.Second).Send(ctrl.GetDefaultSender()); err != nil {
				log.WithField("faultDestRouterId", router.Id).WithError(err).
					Error("failed to send link fault to router on link removal")
			} else {
				log.WithField("faultDestRouterId", router.Id).
					Info("sent link fault to router on link removal")
			}
		}
	}
}

func (network *Network) rerouteLink(l *model.Link, deadline time.Time) error {
	circuits := network.Circuit.All()
	for _, circuit := range circuits {
		if circuit.Path.UsesLink(l) {
			log := logrus.WithField("linkId", l.Id).
				WithField("circuitId", circuit.Id)
			log.Info("circuit uses link")
			if err := network.rerouteCircuit(circuit, deadline); err != nil {
				log.WithError(err).Error("error rerouting circuit, removing")
				if err := network.RemoveCircuit(circuit.Id, true); err != nil {
					log.WithError(err).Error("error removing circuit after reroute failure")
				}
			}
		}
	}

	return nil
}

func (network *Network) rerouteCircuitWithTries(circuit *model.Circuit, retries int) bool {
	log := pfxlog.Logger().WithField("circuitId", circuit.Id)

	// Path is nil for reserved circuits that haven't been built yet
	if circuit.Path.IsValid() {
		for i := 0; i < retries; i++ {
			deadline := time.Now().Add(config.DefaultOptionsRouteTimeout)
			err := network.rerouteCircuit(circuit, deadline)
			if err == nil {
				return true
			}

			log.WithError(err).WithField("attempt", i).Error("error re-routing circuit")
		}
	}

	if err := network.RemoveCircuit(circuit.Id, true); err != nil {
		log.WithError(err).Error("failure while removing circuit after failed re-route attempt")
	}
	return false
}

func (network *Network) rerouteCircuit(circuit *model.Circuit, deadline time.Time) error {
	log := pfxlog.Logger().WithField("circuitId", circuit.Id)
	if circuit.Rerouting.CompareAndSwap(false, true) {
		defer circuit.Rerouting.Store(false)

		log.Warn("rerouting circuit")

		oldPath := circuit.Path
		if cq, err := network.UpdatePath(circuit.Path); err == nil {
			circuit.Path = cq
			circuit.UpdatedAt = time.Now()

			rms := network.CreateRouteMessages(cq, SmartRerouteAttempt, circuit.Id, circuit.Terminator, deadline)

			for i := 0; i < len(cq.Nodes); i++ {
				if _, err := sendRoute(cq.Nodes[i], rms[i], network.options.RouteTimeout); err != nil {
					log.WithError(err).Errorf("error sending route to [r/%s]", cq.Nodes[i].Id)
					return err
				}
			}

			network.unrouteRemovedPathNodes(log, circuit.Id, oldPath, cq)

			log.Info("rerouted circuit")

			network.CircuitEvent(event.CircuitUpdated, circuit, nil)
			return nil
		} else {
			return err
		}
	} else {
		log.Info("not rerouting circuit, already in progress")
		return nil
	}
}

func (network *Network) smartReroute(circuit *model.Circuit, cq *model.Path, deadline time.Time) bool {
	retry := false
	log := pfxlog.Logger().WithField("circuitId", circuit.Id)
	if circuit.Rerouting.CompareAndSwap(false, true) {
		defer circuit.Rerouting.Store(false)

		oldPath := circuit.Path
		circuit.Path = cq
		circuit.UpdatedAt = time.Now()

		rms := network.CreateRouteMessages(cq, SmartRerouteAttempt, circuit.Id, circuit.Terminator, deadline)

		for i := 0; i < len(cq.Nodes); i++ {
			if _, err := sendRoute(cq.Nodes[i], rms[i], network.options.RouteTimeout); err != nil {
				retry = true
				log.WithField("routerId", cq.Nodes[i].Id).WithError(err).Error("error sending smart route update to router")
				break
			}
		}

		if !retry {
			network.unrouteRemovedPathNodes(log, circuit.Id, oldPath, cq)
			logrus.Debug("rerouted circuit")
			network.CircuitEvent(event.CircuitUpdated, circuit, nil)
		}
	}
	return retry
}

func (network *Network) AcceptMetricsMsg(metrics *metrics_pb.MetricsMessage) {
	if metrics.SourceId == network.nodeId {
		return // ignore metrics coming from the controller itself
	}

	log := pfxlog.Logger()

	router, err := network.Router.Read(metrics.SourceId)
	if err != nil {
		log.Debugf("could not find router [r/%s] while processing metrics", metrics.SourceId)
		return
	}

	for _, link := range network.GetAllLinksForRouter(router.Id) {
		metricId := "link." + link.Id + ".latency"
		var latencyCost int64
		var found bool
		if latency, ok := metrics.Histograms[metricId]; ok {
			latencyCost = int64(latency.Mean)
			found = true

			metricId = "link." + link.Id + ".queue_time"
			if queueTime, ok := metrics.Histograms[metricId]; ok {
				latencyCost += int64(queueTime.Mean)
			}
		}

		if found {
			if link.Src.Id == router.Id {
				link.SetSrcLatency(latencyCost) // latency is in nanoseconds
			} else if link.DstId == router.Id {
				link.SetDstLatency(latencyCost) // latency is in nanoseconds
			} else {
				log.Warnf("link not for router")
			}
		}
	}
}

func sendRoute(r *model.Router, createMsg *ctrl_pb.Route, timeout time.Duration) (xt.PeerData, error) {
	log := pfxlog.Logger().WithField("routerId", r.Id).
		WithField("circuitId", createMsg.CircuitId)

	log.Debug("sending create route message")

	msg, err := protobufs.MarshalTyped(createMsg).WithTimeout(timeout).SendForReply(r.Control.GetHighPrioritySender())
	if err != nil {
		log.WithError(err).WithField("timeout", timeout).Error("error sending route message")
		return nil, err
	}

	if msg.ContentType == ctrl_msg.RouteResultType {
		_, success := msg.Headers[ctrl_msg.RouteResultSuccessHeader]
		if !success {
			message := "route error, but no error message from router"
			if errMsg, found := msg.Headers[ctrl_msg.RouteResultErrorHeader]; found {
				message = string(errMsg)
			}
			return nil, errors.New(message)
		}

		peerData := xt.PeerData{}
		for k, v := range msg.Headers {
			if k > 0 {
				peerData[uint32(k)] = v
			}
		}

		return peerData, nil
	}
	return nil, fmt.Errorf("unexpected response type %v received in reply to route request", msg.ContentType)
}

func sendUnroute(r *model.Router, circuitId string, now bool) error {
	unroute := &ctrl_pb.Unroute{
		CircuitId: circuitId,
		Now:       now,
	}
	return protobufs.MarshalTyped(unroute).Send(r.Control.GetHighPrioritySender())
}

// unrouteRemovedPathNodes sends Unroute to any nodes that were in the old path but are not in the new path.
// This handles the case where a circuit reroute changes intermediate transit nodes.
func (network *Network) unrouteRemovedPathNodes(log *logrus.Entry, circuitId string, oldPath, newPath *model.Path) {
	newNodeIds := make(map[string]struct{}, len(newPath.Nodes))
	for _, r := range newPath.Nodes {
		newNodeIds[r.Id] = struct{}{}
	}

	for _, r := range oldPath.Nodes {
		if _, ok := newNodeIds[r.Id]; !ok {
			if cr := network.GetConnectedRouter(r.Id); cr != nil {
				if err := sendUnroute(cr, circuitId, true); err != nil {
					log.WithError(err).Errorf("error sending unroute to removed path node [r/%s]", r.Id)
				}
			}
		}
	}
}

func (network *Network) showOptions() {
	if jsonOptions, err := json.MarshalIndent(network.options, "", "  "); err == nil {
		pfxlog.Logger().Infof("network = %s", string(jsonOptions))
	} else {
		panic(err)
	}
}

type renderConfig interface {
	RenderJsonConfig() (string, error)
}

func (network *Network) routerDeleted(routerId string) {
	circuits := network.GetAllCircuits()
	for _, circuit := range circuits {
		if circuit.HasRouter(routerId) {
			path := circuit.Path

			// If we're either the initiator, terminator (or both), cleanup the circuit since
			// we won't be able to re-establish it, and we'll never get a circuit fault
			if path.Nodes[0].Id == routerId || path.Nodes[len(path.Nodes)-1].Id == routerId {
				if err := network.RemoveCircuit(circuit.Id, true); err != nil {
					pfxlog.Logger().WithField("routerId", routerId).
						WithField("circuitId", circuit.Id).
						WithError(err).Error("unable to remove circuit after router was deleted")
				}
			}
		}
	}
}

var DbSnapshotTooFrequentError = dbSnapshotTooFrequentError{}

type dbSnapshotTooFrequentError struct{}

func (d dbSnapshotTooFrequentError) Error() string {
	return "may snapshot database at most once per minute"
}

func (network *Network) SnapshotDatabase() error {
	_, err := network.SnapshotDatabaseToFile("")
	return err
}

func (network *Network) SnapshotDatabaseToFile(path string) (string, error) {
	network.lock.Lock()
	defer network.lock.Unlock()

	if network.lastSnapshot.Add(time.Minute).After(time.Now()) {
		return "", DbSnapshotTooFrequentError
	}

	actualPath := path
	if actualPath == "" {
		actualPath = "__DB_DIR__/__DB_FILE__-__DATE__-__TIME__"
	}

	err := network.GetDb().View(func(tx *bbolt.Tx) error {
		currentIndex := db.LoadCurrentRaftIndex(tx)
		actualPath = strings.ReplaceAll(actualPath, "__RAFT_INDEX__", fmt.Sprintf("%v", currentIndex))
		actualPath = strings.ReplaceAll(actualPath, "RAFT_INDEX", fmt.Sprintf("%v", currentIndex))

		var err error
		actualPath, _, err = network.GetDb().SnapshotInTx(tx, actualPath)
		return err
	})

	if err == nil {
		network.lastSnapshot = time.Now()
	}

	return actualPath, err
}

func (network *Network) RestoreSnapshot(cmd *command.SyncSnapshotCommand, index uint64) error {
	log := pfxlog.Logger()
	currentTimelineId, err := network.GetDb().GetTimelineId(boltz.TimelineModeDefault, shortid.Generate)
	if err != nil {
		log.WithError(err).Error("unable to get current timeline id")
	}
	if currentTimelineId != "" && currentTimelineId == cmd.TimelineId {
		log.WithField("timelineId", cmd.TimelineId).Info("snapshot already current, skipping reload")
		// DB already restored; ensure cluster id then raft index (index last, see main path).
		if err = network.ensureClusterId(cmd.ClusterId); err != nil {
			return fmt.Errorf("failed to set cluster id for already-current snapshot (%w)", err)
		}
		if err = network.ensureRaftIndex(index); err != nil {
			return fmt.Errorf("failed to set raft index for already-current snapshot (%w)", err)
		}
		return nil
	}

	buf := bytes.NewBuffer(cmd.Snapshot)
	reader, err := gzip.NewReader(buf)
	if err != nil {
		return fmt.Errorf("unable to create gz reader for reading migration snapshot during restore (%w)", err)
	}

	network.GetDb().RestoreFromReader(reader)

	// Write the cluster id before the raft index. The index is the completion gate on restart (the
	// FSM skips entries at or below the stored index), so persist it last: any earlier failure then
	// replays and retries instead of skipping the command with a blank cluster id.
	if err = network.ensureClusterId(cmd.ClusterId); err != nil {
		return fmt.Errorf("failed to set cluster id after db restore (%w)", err)
	}
	if err = network.ensureRaftIndex(index); err != nil {
		return fmt.Errorf("failed to set raft index after db restore (%w)", err)
	}

	time.AfterFunc(5*time.Second, func() {
		if network.restartSelfOnSnapshot {
			log.Info("database restore requires controller restart, restarting...")
			// RestartController returns only when the restart failed; on success it replaces this
			// process and never comes back.
			err := raft.RestartController()
			log.WithError(err).Error("failed to restart controller after snapshot restore, exiting...")
		} else {
			log.Info("database restore requires controller restart. exiting...")
		}
		os.Exit(0)
	})

	return nil
}

// ensureClusterId writes the cluster id if one is not already set (no-op when empty or matching).
// The snapshot restore replaces the whole db, which carries no cluster id, so it must be set here.
func (network *Network) ensureClusterId(clusterId string) error {
	if clusterId == "" {
		return nil
	}
	_, err := db.InitClusterId(network.GetDb(), nil, clusterId)
	return err
}

// ensureRaftIndex records the raft index if the stored one is behind it, returning an error on
// failure so a missed index update surfaces rather than being treated as success.
func (network *Network) ensureRaftIndex(index uint64) error {
	return network.GetDb().Update(nil, func(ctx boltz.MutateContext) error {
		if db.LoadCurrentRaftIndex(ctx.Tx()) >= index {
			return nil
		}
		raftBucket := boltz.GetOrCreatePath(ctx.Tx(), db.RootBucket, db.MetadataBucket)
		raftBucket.SetInt64(db.FieldRaftIndex, int64(index), nil)
		return raftBucket.GetError()
	})
}

// SetRestartSelfOnSnapshot controls whether the controller restarts itself after a snapshot restore
// (true) or exits expecting an external restart (false). It mirrors the raft restartSelfOnSnapshot
// setting and is applied by the owning controller after construction.
func (network *Network) SetRestartSelfOnSnapshot(v bool) {
	network.restartSelfOnSnapshot = v
}

func (network *Network) AddInspectTarget(target InspectTarget) {
	network.inspectionTargets.Append(target)
}

// SetCtrlDialerValidator registers the ctrl dialer's validation function with the network.
func (network *Network) SetCtrlDialerValidator(validator CtrlDialerValidator) {
	network.ctrlDialerValidator = validator
}

// GetCtrlDialerValidator returns the registered ctrl dialer validation function, or nil.
func (network *Network) GetCtrlDialerValidator() CtrlDialerValidator {
	return network.ctrlDialerValidator
}

func (network *Network) ValidateRouterLinks(router *model.Router, cb LinkValidationCallback) {
	request := &ctrl_pb.InspectRequest{RequestedValues: []string{"links"}}
	resp := &ctrl_pb.InspectResponse{}
	respMsg, err := protobufs.MarshalTyped(request).WithTimeout(time.Minute).SendForReply(router.Control.GetDefaultSender())
	if err = protobufs.TypedResponse(resp).Unmarshall(respMsg, err); err != nil {
		network.reportRouterLinksError(router, err, cb)
		return
	}

	var linkDetails *inspect.LinksInspectResult
	for _, val := range resp.Values {
		if val.Name == "links" {
			if err = json.Unmarshal([]byte(val.Value), &linkDetails); err != nil {
				network.reportRouterLinksError(router, err, cb)
				return
			}
		}
	}

	if linkDetails == nil {
		if len(resp.Errors) > 0 {
			err = errors.New(strings.Join(resp.Errors, ","))
			network.reportRouterLinksError(router, err, cb)
			return
		}
		network.reportRouterLinksError(router, errors.New("no link details returned from router"), cb)
		return
	}

	linkMap := network.Link.GetLinkMap()

	result := &mgmt_pb.RouterLinkDetails{
		RouterId:        router.Id,
		RouterName:      router.Name,
		ValidateSuccess: true,
	}

	for _, link := range linkDetails.Links {
		detail := &mgmt_pb.RouterLinkDetail{
			LinkId:       link.Id,
			RouterState:  mgmt_pb.LinkState_LinkEstablished,
			DestRouterId: link.Dest,
			Dialed:       link.Dialed,
		}
		detail.DestConnected = network.ConnectedRouter(link.Dest)
		if ctrlLink, found := linkMap[link.Id]; found {
			detail.CtrlState = mgmt_pb.LinkState_LinkEstablished
			detail.IsValid = detail.DestConnected
			if link.Dialed { // only compare against dialer side of the link, as src/dst will be flipped on the listener side
				network.checkLinkConns(ctrlLink, link, detail)
			}
		} else {
			detail.CtrlState = mgmt_pb.LinkState_LinkUnknown
			detail.IsValid = !detail.DestConnected
		}
		delete(linkMap, link.Id)
		result.LinkDetails = append(result.LinkDetails, detail)
	}

	for _, link := range linkMap {
		related := false
		dest := ""
		if link.Src.Id == router.Id {
			related = true
			dest = link.DstId
		} else if link.DstId == router.Id {
			related = true
			dest = link.Src.Id
		}

		if related {
			detail := &mgmt_pb.RouterLinkDetail{
				LinkId:        link.Id,
				CtrlState:     mgmt_pb.LinkState_LinkEstablished,
				DestConnected: network.ConnectedRouter(dest),
				RouterState:   mgmt_pb.LinkState_LinkUnknown,
				IsValid:       false,
				DestRouterId:  dest,
				Dialed:        link.Src.Id == router.Id,
			}
			result.LinkDetails = append(result.LinkDetails, detail)
		}
	}

	cb(result)
}

func (network *Network) checkLinkConns(ctrlLink *model.Link, routerLink *inspect.LinkInspectDetail, result *mgmt_pb.RouterLinkDetail) {
	sortF := func(v []*ctrl_pb.LinkConn) []*ctrl_pb.LinkConn {
		return slices.SortedFunc(slices.Values(v), func(e *ctrl_pb.LinkConn, e2 *ctrl_pb.LinkConn) int {
			if diff := strings.Compare(e.Type, e2.Type); diff != 0 {
				return diff
			}
			if diff := strings.Compare(e.LocalAddr, e2.LocalAddr); diff != 0 {
				return diff
			}
			return strings.Compare(e.RemoteAddr, e2.RemoteAddr)
		})
	}

	var routerConns []*ctrl_pb.LinkConn

	for _, v := range routerLink.Connections {
		routerConns = append(routerConns, &ctrl_pb.LinkConn{
			Type:       v.Type,
			LocalAddr:  v.Source,
			RemoteAddr: v.Dest,
		})
	}

	// The version comes from the hello, so only the connected instance has it. Not knowing it means the
	// comparison cannot be made, which is not a fault of the link.
	if srcR := ctrlLink.Src; srcR != nil {
		connectedSrc := network.Router.GetConnected(srcR.Id)
		if connectedSrc == nil || connectedSrc.VersionInfo == nil {
			return
		}
		hasMinVersion, err := connectedSrc.VersionInfo.HasMinimumVersion("v1.6.6")
		if err != nil {
			result.IsValid = false
			result.Messages = append(result.Messages, err.Error())
			return
		}
		if !hasMinVersion {
			return
		}
	}

	ctrlConnState := ctrlLink.GetConnsState()
	ctrlConns := sortF(ctrlConnState.GetConns())
	routerConns = sortF(routerConns)

	if len(ctrlConns) != len(routerConns) {
		result.Messages = append(result.Messages, fmt.Sprintf("for link %s, len(ctrlConns): %d != len(routerConns): %d",
			ctrlLink.Id, len(ctrlConns), len(ctrlConns)))
		return
	}

	if routerLink.ConnStateIteration != ctrlConnState.GetStateIteration() {
		result.Messages = append(result.Messages, fmt.Sprintf("for link %s, conn state iteration doesn't match, ctrl: %d != router: %d",
			ctrlLink.Id, ctrlConnState.GetStateIteration(), routerLink.ConnStateIteration))
		return
	}

	for i := 0; i < len(ctrlConns); i++ {
		ctrlConn := ctrlConns[i]
		routerConn := routerConns[i]
		if ctrlConn.Type != routerConn.Type {
			result.IsValid = false
			result.Messages = append(result.Messages, fmt.Sprintf("for link %s, type doesn't match. ctrl %s != router %s",
				ctrlLink.Id, ctrlConn.Type, routerConn.Type))
		}
		if ctrlConn.LocalAddr != routerConn.LocalAddr {
			result.IsValid = false
			result.Messages = append(result.Messages, fmt.Sprintf("for link %s, local addr doesn't match. ctrl %s != router %s",
				ctrlLink.Id, ctrlConn.LocalAddr, routerConn.LocalAddr))
		}
		if ctrlConn.RemoteAddr != routerConn.RemoteAddr {
			result.IsValid = false
			result.Messages = append(result.Messages, fmt.Sprintf("for link %s, remote addr doesn't match. ctrl %s != router %s",
				ctrlLink.Id, ctrlConn.RemoteAddr, routerConn.RemoteAddr))
		}
	}
}

func (network *Network) reportRouterLinksError(router *model.Router, err error, cb LinkValidationCallback) {
	result := &mgmt_pb.RouterLinkDetails{
		RouterId:        router.Id,
		RouterName:      router.Name,
		ValidateSuccess: false,
		Message:         err.Error(),
	}
	cb(result)
}

func minCost(q map[*model.Router]bool, dist map[*model.Router]int64) *model.Router {
	if len(dist) < 1 {
		return nil
	}

	currentMin := int64(math.MaxInt64)
	var selected *model.Router
	for r := range q {
		d := dist[r]
		if d <= currentMin {
			selected = r
			currentMin = d
		}
	}
	return selected
}

type Cache interface {
	RemoveFromCache(id string)
}

func newPathAndCost(path []*model.Router, cost int64) *PathAndCost {
	if cost > (1 << 20) {
		cost = 1 << 20
	}
	return &PathAndCost{
		path: path,
		cost: uint32(cost),
	}
}

type PathAndCost struct {
	path []*model.Router
	cost uint32
}

type InvalidCircuitError struct {
	circuitId string
}

func (err InvalidCircuitError) Error() string {
	return fmt.Sprintf("invalid circuit (%s)", err.circuitId)
}