Files
ziti/controller/model/router_model.go
Paul Lorenz 647c4daa1e Serialize router control-channel connect/disconnect. Fixes #4196
The controller decided which of two racing connections for a router was current
by comparing router instances, but loaded one per connect by evicting the router
cache and reading back through it. Two connects could both evict, and whichever
read second was handed the instance the first had just published. A shared
instance makes the two connections indistinguishable: the connect path cannot
reject the second into an occupied slot, and when either channel dies the
disconnect path finds itself current and tears down the registration the other is
still using. The surviving channel is never re-bound, so the router stays
connected at the transport layer while absent from the model, unable to recover.

Connect and disconnect were also unserialized, so a stale or superseded
disconnect could interleave with a live connection and take its links with it.

- serializes a router's connect and disconnect with a per-router striped lock
- keeps at most one connection per router: a connect into an occupied slot is
  rejected via an error from ConnectRouter, so the bind fails and NewChannel
  closes it without starting rx or registering it, and the occupant is displaced;
  the router redials into the freed slot
- displaces an occupant by closing its channel and also invoking the teardown
  directly, since a channel that is already closed never fires its close handler
  again; without this a dead but still registered connection holds the slot
  forever and every redial is rejected against a slot nothing can free
- refuses a connect whose control channel is already closed rather than
  registering it, so a connection no disconnect could ever remove is never
  published
- gives every connection its own router instance via RouterManager.NewCtrlChanRouter,
  read through readUncached so the cache neither supplies nor receives it, which
  is what makes comparing instances meaningful
- moves recording the channel and connect time out of the accept path, so a
  caller cannot attach the wrong channel or forget to attach one
- serializes link publication with that teardown on the same per-router stripe.
  Validating currency and then publishing without it is a check-then-act: a
  report can find the connection current and, by the time it reaches the link
  manager, the teardown has already snapshotted and cleared the router's links,
  so the link is recreated after everything that would have removed it. It is
  then absent from the router's own index while still in the link table with a
  disconnected source, and a reconnect reporting the same iteration can adopt
  that stale source instead of rebuilding the link
- guards the entire DisconnectRouter teardown by connection currency, all or
  nothing, and clears the connected flag and link index only when the
  registration was actually given up, with the flag cleared under the same shard
  lock as the map removal so the two cannot be observed disagreeing; the connected flag decides whether the
  controller accepts a router's link reports, so clearing it for the wrong
  connection silences a router that is up and reporting
- reduces MarkConnected to publishing the connection; the takeover-close moves
  into ConnectRouter's reject path
- makes the per-router unlock idempotent so callers can defer it as a leak-safety
  net and still unlock early before closing a channel outside the lock
- stops the replaced RouterSender in routerTxMap.Add so a takeover does not leak
  the old sender's goroutine when the broker's asynchronous RouterDisconnected
  loses the race to the redial's RouterConnected
- discards pending peer state changes for a router whose channel has closed,
  since sending on one fails immediately and the failed send is retried as soon
  as the event loop turns, spinning the loop and flooding the log
- queues the peer-state send-done event on every path, so a missing channel can
  no longer leave sendInProgress set and stall that router's updates permanently
- resolves a router's version from its connected instance when validating link
  conn info, since the version arrives in the hello and so is absent from an
  instance loaded from the database
- normalizes both endpoints to the connected instance in shortestPath, which is
  keyed and compared by pointer and so treated an endpoint held as any other
  instance of the same router as absent from the graph, reporting a router as
  unroutable from itself. That worked before only because the connect path
  published its instance into the router cache, so a cache read and the connected
  map returned the same object; nothing stated the requirement
- configures test logging once per package in TestMain, so a test no longer
  writes global logger state while a previous test's shutdown logging reads it
2026-08-13 23:54:40 -04:00

189 lines
6.4 KiB
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 model
import (
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/openziti/foundation/v2/errorz"
"github.com/openziti/foundation/v2/versions"
"github.com/openziti/ziti/v2/common/capabilities"
"github.com/openziti/ziti/v2/common/ctrlchan"
"github.com/openziti/ziti/v2/common/pb/ctrl_pb"
"github.com/openziti/ziti/v2/controller/db"
"github.com/openziti/ziti/v2/controller/models"
"github.com/openziti/ziti/v2/controller/storage/boltz"
"go.etcd.io/bbolt"
)
type Router struct {
models.BaseEntity
Name string
Fingerprint *string
// listeners is the router's currently-advertised link listener set.
// Hello carries the initial snapshot; UpdateLinkListeners pushes
// mid-session changes from the router. Guarded by mu, so it is
// unexported: reach it through SetLinkListeners / GetLinkListeners.
listeners []*ctrl_pb.Listener
// mu guards the fields a connected Router mutates mid-session, which
// today is just listeners. Fields added later should share it rather
// than take their own lock; contention is irrelevant at this scale.
mu sync.RWMutex
Control ctrlchan.CtrlChannel
Connected atomic.Bool
ConnectTime time.Time
// VersionInfo is reported in the router's hello and is not persisted, so it is only populated on the
// instance built for a control-channel connection. It is nil on an instance loaded from the database.
// Read it from GetConnected rather than from whatever instance is to hand.
VersionInfo *versions.VersionInfo
routerLinks RouterLinks
Cost uint16
NoTraversal bool
Disabled bool
Capabilities *capabilities.RouterCapabilityMask
Interfaces []*Interface
CtrlChanListeners map[string][]string
Configs []string
}
func (entity *Router) GetLinks() []*Link {
return entity.routerLinks.GetLinks()
}
func (entity *Router) toBoltEntityForUpdate(tx *bbolt.Tx, env Env, checker boltz.FieldChecker) (*db.Router, error) {
if err := validateRouterConfigs(tx, env, entity.Configs, checker); err != nil {
return nil, err
}
return entity.toBoltEntity(), nil
}
func (entity *Router) toBoltEntityForCreate(tx *bbolt.Tx, env Env) (*db.Router, error) {
if err := validateRouterConfigs(tx, env, entity.Configs, nil); err != nil {
return nil, err
}
return entity.toBoltEntity(), nil
}
func (entity *Router) toBoltEntity() *db.Router {
return &db.Router{
BaseExtEntity: *boltz.NewExtEntity(entity.Id, entity.Tags),
Name: entity.Name,
Fingerprint: entity.Fingerprint,
Cost: entity.Cost,
NoTraversal: entity.NoTraversal,
Disabled: entity.Disabled,
CtrlChanListeners: entity.CtrlChanListeners,
Interfaces: InterfacesToBolt(entity.Interfaces),
Configs: entity.Configs,
}
}
// validateRouterConfigs validates that all configs target routers and have unique config types.
// If checker is non-nil and the configs field is not being updated, validation is skipped.
func validateRouterConfigs(tx *bbolt.Tx, env Env, configs []string, checker boltz.FieldChecker) error {
if checker != nil && !checker.IsUpdated(db.EntityTypeConfigs) {
return nil
}
if len(configs) == 0 {
return nil
}
typeMap := map[string]*db.Config{}
configStore := env.GetStores().Config
configTypeStore := env.GetStores().ConfigType
for _, id := range configs {
config, err := configStore.LoadById(tx, id)
if err != nil {
return err
}
configType, err := configTypeStore.LoadById(tx, config.TypeId)
if err != nil {
if boltz.IsErrNotFoundErr(err) {
msg := fmt.Sprintf("config %v references config type %v which does not exist",
config.Name, config.TypeId)
return errorz.NewFieldError(msg, "configs", configs)
}
return err
}
if configType.Target != db.ConfigTypeTargetRouter {
msg := fmt.Sprintf("config %v has config type %v which does not target routers",
config.Name, configType.Name)
return errorz.NewFieldError(msg, "configs", configs)
}
if conflictConfig, found := typeMap[config.TypeId]; found {
msg := fmt.Sprintf("duplicate configs named %v and %v found for config type %v. Only one config of a given type is allowed per router",
conflictConfig.Name, config.Name, configType.Name)
return errorz.NewFieldError(msg, "configs", configs)
}
typeMap[config.TypeId] = config
}
return nil
}
func (entity *Router) fillFrom(_ Env, _ *bbolt.Tx, boltRouter *db.Router) error {
entity.Name = boltRouter.Name
entity.Fingerprint = boltRouter.Fingerprint
entity.Cost = boltRouter.Cost
entity.NoTraversal = boltRouter.NoTraversal
entity.Disabled = boltRouter.Disabled
entity.CtrlChanListeners = boltRouter.CtrlChanListeners
entity.Interfaces = InterfacesFromBolt(boltRouter.Interfaces)
entity.Configs = boltRouter.Configs
entity.FillCommon(boltRouter)
return nil
}
func (entity *Router) addLinkListener(addr, linkProtocol string, groups []string) {
entity.mu.Lock()
defer entity.mu.Unlock()
entity.listeners = append(entity.listeners, &ctrl_pb.Listener{
Address: addr,
Protocol: linkProtocol,
Groups: groups,
})
}
// SetLinkListeners atomically replaces the router's link listener slice.
// Callers do not mutate the previous slice — readers may still hold and
// iterate it safely after a Set.
func (entity *Router) SetLinkListeners(listeners []*ctrl_pb.Listener) {
entity.mu.Lock()
defer entity.mu.Unlock()
entity.listeners = listeners
}
// GetLinkListeners returns the current link listener slice under a read
// lock. The returned slice is the live header — safe to iterate because
// SetLinkListeners always replaces the whole slice rather than mutating
// it in place — but callers should not modify it.
func (entity *Router) GetLinkListeners() []*ctrl_pb.Listener {
entity.mu.RLock()
defer entity.mu.RUnlock()
return entity.listeners
}
func (entity *Router) HasCapability(capability capabilities.RouterCapability) bool {
return entity.Capabilities != nil && entity.Capabilities.IsSet(capability)
}