ip_remote_multi_client.go

package connect

import (
	"context"
	"sync"
	"sync/atomic"
	"time"

	// "reflect"
	"errors"
	"fmt"
	"math"
	mathrand "math/rand"
	"slices"
	"strings"

	"golang.org/x/exp/maps"
	"golang.org/x/net/publicsuffix"

	"google.golang.org/protobuf/proto"

	"github.com/urnetwork/glog"

	"github.com/urnetwork/connect/protocol"
)

// multi client is a sender approach to mitigate bad destinations
// it maintains a window of compatible clients chosen using specs
// (e.g. from a desription of the intent of use)
// - the clients are rate limited by the number of outstanding acks (nacks)
// - the size of allowed outstanding nacks increases with each ack,
// scaling up successful destinations to use the full transfer buffer
// - the clients are chosen with probability weighted by their
// net frame count statistics (acks - nacks)

// the following functions handle moving clients in and out of the window:
// - `resize`
//   The goal of the resize is to meet a target window size based on the number
//   of different source ip:port per destination ip:port.
//   Two statistics are used: effective bytes per second ([ack used])
//     and expected bytes per second ([capacity]-[ack used]-[unacked used]).
//   Unhealthy clients are removed from the window based on low effective stats,
//   unless the window is fixed size.
//   Fundamentally this approach can't tell the difference between an
//   unhealthy client and an idle client, so the norm is to continually change clients
//   in the lull after a burst of usage.
// - `detectBlackhole`
//   When a client acks traffic but does not return traffic,
//   it gets labeled a black hole. Black hole clients may be malicious
//   or have network filtering. Black hole clients are removed.
// - `ping`
//   When a client is idle it must continually ack ping requests.
//   Clients that fail to ack are removed.

// TODO surface window stats to show to users

type clientReceivePacketFunction func(client *multiClientChannel, source TransferPath, provideMode protocol.ProvideMode, ipPath *IpPath, packet []byte)

type DestinationStats struct {
	EstimatedBytesPerSecond ByteCount
	Tier                    int
}

type WindowType int

func (self WindowType) RankMode() string {
	switch self {
	case WindowTypeQuality:
		return "quality"
	case WindowTypeSpeed:
		return "speed"
	default:
		return ""
	}
}

const (
	WindowTypeQuality WindowType = 0
	WindowTypeSpeed   WindowType = 1
)

// for each `NewClientArgs`,
//
//	`RemoveClientWithArgs` will be called if a client was created for the args,
//	else `RemoveClientArgs`
type MultiClientGenerator interface {
	// path -> estimated byte count per second
	// the enumeration should typically
	// 1. not repeat final destination ids from any path
	// 2. not repeat intermediary elements from any path
	NextDestinations(count int, excludeDestinations []MultiHopId, rankMode string) (map[MultiHopId]DestinationStats, error)
	// client id, client auth
	NewClientArgs() (*MultiClientGeneratorClientArgs, error)
	RemoveClientArgs(args *MultiClientGeneratorClientArgs)
	RemoveClientWithArgs(client *Client, args *MultiClientGeneratorClientArgs)
	NewClientSettings() *ClientSettings
	NewClient(ctx context.Context, args *MultiClientGeneratorClientArgs, clientSettings *ClientSettings) (*Client, error)
	FixedDestinationSize() (int, bool)
}

func DefaultMultiClientSettings() *MultiClientSettings {
	return &MultiClientSettings{
		SequenceIdleTimeout: 120 * time.Second,

		WindowSizes: map[WindowType]WindowSizeSettings{
			// TODO increase `WindowSizeMinP2pOnly` when p2p is deployed
			WindowTypeQuality: WindowSizeSettings{
				WindowSizeMin:     2,
				WindowSizeMax:     6,
				WindowSizeHardMax: 12,
				// reconnects per source
				WindowSizeReconnectScale: 1.2,
				KeepHealthiestCount:      0,
			},
			WindowTypeSpeed: WindowSizeSettings{
				WindowSizeMin:     1,
				WindowSizeMax:     2,
				WindowSizeHardMax: 4,
				FixedWindowSize:   1,
				// WindowSizeUseMax:     1,
				// reconnects per source
				WindowSizeReconnectScale: 1.0,
				KeepHealthiestCount:      0,
			},
		},
		SendRetryTimeout:           2000 * time.Millisecond,
		PingWriteTimeout:           5 * time.Second,
		CPingWriteTimeout:          15 * time.Second,
		CPingMaxByteCountPerSecond: kib(32),
		// the initial ping includes creating the transports and contract
		// ease up the timeout until perf issues are fully resolved
		PingTimeout:  30 * time.Second,
		CPingTimeout: 30 * time.Second,
		// a lower ack timeout helps cycle through bad providers faster
		AckTimeout:                                30 * time.Second,
		BlackholeTimeout:                          5 * time.Second,
		BlackholeConnectTimeout:                   30 * time.Second,
		WindowResizeTimeout:                       15 * time.Second,
		StatsWindowGraceperiod:                    30 * time.Second,
		StatsWindowMaxEstimatedByteCountPerSecond: mib(16),
		// StatsWindowMaxEffectiveByteCountPerSecondScale: 0.8,
		StatsWindowEntropy:  0.0,
		WindowExpandTimeout: 15 * time.Second,
		// WindowExpandBlockTimeout: 5 * time.Second,
		WindowExpandBlockCount: 4,
		// wait this time before enumerating potential clients again
		// WindowEnumerateEmptyTimeout: 5 * time.Second,
		WindowEnumerateErrorTimeout: 1 * time.Second,
		// WindowMaxScale:              4.0,
		// WindowExpandMaxOvershotScale: 2.0,
		// WindowRevisitTimeout:      2 * time.Minute,
		WindowExpandArgsTimeout:   2 * time.Minute,
		StatsWindowDuration:       30 * time.Second,
		StatsWindowBucketDuration: 1 * time.Second,
		StatsSampleWeightsCount:   8,
		// percentile
		StatsSourceCountSelection: 0.9,
		// ClientAffinityTimeout:        0 * time.Second,

		MultiRaceSetOnNoResponseTimeout:      5 * time.Second,
		MultiRaceSetOnResponseTimeout:        2 * time.Second,
		MultiRaceClientSentPacketMaxCount:    16,
		MultiRaceClientPacketMaxCount:        8,
		MultiRacePacketMaxCount:              32,
		MultiRaceClientEarlyCompleteFraction: 0.25,
		// TODO on platforms with more memory, increase this
		MultiRaceClientCount: 0,

		StatsWindowMaxUnhealthyDuration:  15 * time.Second,
		StatsWindowWarnUnhealthyDuration: 5 * time.Second,
		// StatsWindowKeepHealthiestCount:   2,
		// the effective byte count is per stats window `StatsWindowDuration`
		// StatsWindowMinHealthyEffectiveSendByteCount:    kib(1),
		// StatsWindowMinHealthyEffectiveReceiveByteCount: kib(32),

		MaxClientLifetime: 60 * time.Minute,

		ProtocolVersion:     DefaultProtocolVersion,
		DestinationAffinity: true,

		DefaultReconnectScale: 1.0,
		DefaultUlimit:         0,

		TcpCollapsePrevention: true,
		UdpCollapsePrevention: false,

		IngressSecurityPolicyGenerator: DefaultIngressSecurityPolicyWithStats,
		EgressSecurityPolicyGenerator:  DefaultEgressSecurityPolicyWithStats,

		RemoteUserNatMultiClientMonitorSettings: *DefaultRemoteUserNatMultiClientMonitorSettings(),
	}
}

type MultiClientSettings struct {
	SequenceIdleTimeout time.Duration
	WindowSizes         map[WindowType]WindowSizeSettings
	// ClientNackInitialLimit int
	// ClientNackMaxLimit int
	// ClientNackScale float64
	// ClientWriteTimeout time.Duration
	// SendTimeout time.Duration
	// WriteTimeout time.Duration
	SendRetryTimeout                          time.Duration
	PingWriteTimeout                          time.Duration
	CPingWriteTimeout                         time.Duration
	CPingMaxByteCountPerSecond                ByteCount
	PingTimeout                               time.Duration
	CPingTimeout                              time.Duration
	AckTimeout                                time.Duration
	BlackholeTimeout                          time.Duration
	BlackholeConnectTimeout                   time.Duration
	WindowResizeTimeout                       time.Duration
	StatsWindowGraceperiod                    time.Duration
	StatsWindowMaxEstimatedByteCountPerSecond ByteCount
	// StatsWindowMaxEffectiveByteCountPerSecondScale float32
	StatsWindowEntropy  float32
	WindowExpandTimeout time.Duration
	// WindowExpandBlockTimeout     time.Duration
	WindowExpandBlockCount int
	// WindowEnumerateEmptyTimeout time.Duration
	WindowEnumerateErrorTimeout time.Duration
	// WindowMaxScale              float64
	// WindowExpandMaxOvershotScale float64
	// WindowRevisitTimeout      time.Duration
	WindowExpandArgsTimeout   time.Duration
	StatsWindowDuration       time.Duration
	StatsWindowBucketDuration time.Duration
	StatsSampleWeightsCount   int
	// percentile
	StatsSourceCountSelection float64
	// lower affinity is more private
	// however, there may be some applications that assume the same ip across multiple connections
	// in those cases, we would need some small affinity
	// ClientAffinityTimeout time.Duration

	// time since first send to end the race, if no response
	MultiRaceSetOnNoResponseTimeout time.Duration
	// time after the first response to end the race
	MultiRaceSetOnResponseTimeout        time.Duration
	MultiRaceClientSentPacketMaxCount    int
	MultiRaceClientPacketMaxCount        int
	MultiRacePacketMaxCount              int
	MultiRaceClientEarlyCompleteFraction float32
	MultiRaceClientCount                 int

	StatsWindowMaxUnhealthyDuration  time.Duration
	StatsWindowWarnUnhealthyDuration time.Duration
	// StatsWindowKeepHealthiestCount                 int
	// StatsWindowMinHealthyEffectiveSendByteCount    ByteCount
	// StatsWindowMinHealthyEffectiveReceiveByteCount ByteCount

	// active clients longer than this lifetime will not be forced closed
	// new connections will be routed to new clients
	MaxClientLifetime time.Duration

	ProtocolVersion int

	// note destination affinity will affect retry with different source ports
	// it relies on the performance of the initial race being good enough,
	// and all-or-nothing bad clients where if one destination does not route via a client,
	// all destinations should not route, so that the client can be detected as unhealthy
	DestinationAffinity bool

	DefaultPerformanceProfile *PerformanceProfile

	// used when reconnect scale is not set in a custom performance profile
	DefaultReconnectScale float64
	// used when ulimit is not set in a custom performance profile
	DefaultUlimit int

	TcpCollapsePrevention bool
	UdpCollapsePrevention bool

	IngressSecurityPolicyGenerator func(*SecurityPolicyStatsCollector) SecurityPolicy
	EgressSecurityPolicyGenerator  func(*SecurityPolicyStatsCollector) SecurityPolicy

	RemoteUserNatMultiClientMonitorSettings
}

type WindowSizeSettings struct {
	WindowSizeMin int
	// the minimumum number of items in the windows that must be connected via p2p only
	WindowSizeMinP2pOnly int
	// inclusive
	WindowSizeMax     int
	WindowSizeHardMax int
	// leave 0 to automatically size between `WindowSizeMin` and `WindowSizeMax`
	FixedWindowSize int
	// WindowSizeUseMax     int
	// clients per source (leave 0 for default)
	WindowSizeReconnectScale float64
	KeepHealthiestCount      int
	// (leave 0 for default)
	Ulimit int
}

func (self *WindowSizeSettings) Validate() error {
	if self.WindowSizeMax < self.WindowSizeMin {
		return fmt.Errorf(
			"Window size [%d, %d] invalid. Max must be >= min",
			self.WindowSizeMin,
			self.WindowSizeMax,
		)
	}

	if 0 < self.FixedWindowSize {
		if self.FixedWindowSize < self.WindowSizeMin || self.WindowSizeMax < self.FixedWindowSize {
			return fmt.Errorf(
				"Window size [%d, %d] must include the fixed size =%d",
				self.WindowSizeMin,
				self.WindowSizeMax,
				self.FixedWindowSize,
			)
		}
	}

	return nil
}

// not setting a performance profile will use the default "auto" mode
// which balances traffic across multiple window types with an internal set of profiles
type PerformanceProfile struct {
	WindowType  WindowType
	WindowSize  WindowSizeSettings
	AllowDirect bool
}

func (self *PerformanceProfile) Validate() error {
	err := self.WindowSize.Validate()
	if err != nil {
		return err
	}

	return nil
}

func DefaultWindowSizeSettings() WindowSizeSettings {
	return WindowSizeSettings{
		WindowSizeMin:            1,
		WindowSizeMax:            1,
		WindowSizeHardMax:        4,
		WindowSizeReconnectScale: 1.0,
		KeepHealthiestCount:      1,
	}
}

type receivePacket struct {
	Source      TransferPath
	ProvideMode protocol.ProvideMode
	IpPath      *IpPath
	Packet      []byte
	Pooled      bool
}

type RemoteUserNatMultiClient struct {
	ctx    context.Context
	cancel context.CancelFunc

	generator MultiClientGenerator

	receivePacketCallback ReceivePacketFunction

	settings *MultiClientSettings

	windows map[WindowType]*multiClientWindow
	monitor MultiClientMonitor

	securityPolicyStats   *SecurityPolicyStatsCollector
	securityPolicy        SecurityPolicy
	ingressSecurityPolicy SecurityPolicy

	// the provide mode of the source packets
	// for locally generated packets this is `ProvideMode_Network`
	provideMode protocol.ProvideMode

	stateLock        sync.Mutex
	ip4PathUpdates   map[Ip4Path]*multiClientChannelUpdate
	ip6PathUpdates   map[Ip6Path]*multiClientChannelUpdate
	affinityIp4Paths map[Ip4Path]map[Ip4Path]time.Time
	affinityIp6Paths map[Ip6Path]map[Ip6Path]time.Time
	clientUpdates    map[*multiClientChannel]map[*multiClientChannelUpdate]bool

	performanceProfile  *PerformanceProfile
	localSecurityBypass bool

	localUserNat      *LocalUserNat
	localUserNatUnsub func()
}

func NewRemoteUserNatMultiClientWithDefaults(
	ctx context.Context,
	generator MultiClientGenerator,
	receivePacketCallback ReceivePacketFunction,
	provideMode protocol.ProvideMode,
) *RemoteUserNatMultiClient {
	return NewRemoteUserNatMultiClient(
		ctx,
		generator,
		receivePacketCallback,
		provideMode,
		DefaultMultiClientSettings(),
	)
}

func NewRemoteUserNatMultiClient(
	ctx context.Context,
	generator MultiClientGenerator,
	receivePacketCallback ReceivePacketFunction,
	provideMode protocol.ProvideMode,
	settings *MultiClientSettings,
) *RemoteUserNatMultiClient {
	cancelCtx, cancel := context.WithCancel(ctx)

	securityPolicyStats := DefaultSecurityPolicyStatsCollector()

	localUserNatSettings := DefaultLocalUserNatSettings()
	// no ulimit for local traffic
	localUserNatSettings.UdpBufferSettings.UserLimit = 0
	localUserNatSettings.TcpBufferSettings.UserLimit = 0
	localUserNat := NewLocalUserNat(cancelCtx, "multi local", localUserNatSettings)

	multiClient := &RemoteUserNatMultiClient{
		ctx:                   cancelCtx,
		cancel:                cancel,
		generator:             generator,
		receivePacketCallback: receivePacketCallback,
		settings:              settings,
		windows:               map[WindowType]*multiClientWindow{},
		securityPolicyStats:   securityPolicyStats,
		securityPolicy:        settings.EgressSecurityPolicyGenerator(securityPolicyStats),
		ingressSecurityPolicy: settings.IngressSecurityPolicyGenerator(securityPolicyStats),
		provideMode:           provideMode,
		ip4PathUpdates:        map[Ip4Path]*multiClientChannelUpdate{},
		ip6PathUpdates:        map[Ip6Path]*multiClientChannelUpdate{},
		affinityIp4Paths:      map[Ip4Path]map[Ip4Path]time.Time{},
		affinityIp6Paths:      map[Ip6Path]map[Ip6Path]time.Time{},
		clientUpdates:         map[*multiClientChannel]map[*multiClientChannelUpdate]bool{},
		performanceProfile:    settings.DefaultPerformanceProfile,
		localUserNat:          localUserNat,
	}

	multiClient.windows[WindowTypeQuality] = newMultiClientWindow(
		cancelCtx,
		cancel,
		generator,
		multiClient.clientReceivePacket,
		multiClient.ingressSecurityPolicy,
		multiClient.removeClient,
		WindowTypeQuality,
		settings,
	)
	if _, fixed := generator.FixedDestinationSize(); !fixed {
		multiClient.windows[WindowTypeSpeed] = newMultiClientWindow(
			cancelCtx,
			cancel,
			generator,
			multiClient.clientReceivePacket,
			multiClient.ingressSecurityPolicy,
			multiClient.removeClient,
			WindowTypeSpeed,
			settings,
		)
	}
	// else only keep the quality window for fixed destination

	multiClient.localUserNatUnsub = localUserNat.AddReceivePacketCallback(receivePacketCallback)

	monitors := []MultiClientMonitor{}
	for _, window := range multiClient.windows {
		monitors = append(monitors, window.monitor)
	}
	multiClient.monitor = NewMergedMultiClientMonitor(monitors)

	return multiClient
}

func (self *RemoteUserNatMultiClient) SecurityPolicyStats(reset bool) SecurityPolicyStats {
	return self.securityPolicyStats.Stats(reset)
}

func (self *RemoteUserNatMultiClient) Monitor() MultiClientMonitor {
	return self.monitor
}

func (self *RemoteUserNatMultiClient) AddContractStatusCallback(contractStatusCallback ContractStatusFunction) func() {
	subs := []func(){}
	for _, window := range self.windows {
		sub := window.AddContractStatusCallback(contractStatusCallback)
		subs = append(subs, sub)
	}
	return func() {
		for _, sub := range subs {
			sub()
		}
	}
}

// the performance profile will take effect at the next `resize` iteration
func (self *RemoteUserNatMultiClient) SetPerformanceProfile(performanceProfile *PerformanceProfile) {
	if performanceProfile != nil {
		err := performanceProfile.Validate()
		if err != nil {
			panic(err)
		}
	}

	func() {
		self.stateLock.Lock()
		defer self.stateLock.Unlock()

		self.performanceProfile = performanceProfile
	}()
	for _, window := range self.windows {
		window.SetPerformanceProfile(performanceProfile)
		// reset the window
		window.shuffle()
	}
}

func (self *RemoteUserNatMultiClient) SetLocalSecurityBypass(localSecurityBypass bool) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()
	self.localSecurityBypass = localSecurityBypass
}

func (self *RemoteUserNatMultiClient) LocalSecurityBypass() bool {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()
	return self.localSecurityBypass
}

// ordered by choice descending
func (self *RemoteUserNatMultiClient) selectWindowTypes(sendPacket *parsedPacket) []WindowType {
	// - web traffic is routed to quality providers
	// - all other traffic is routed to speed providers

	if _, fixed := self.generator.FixedDestinationSize(); fixed {
		return []WindowType{WindowTypeQuality}
	}

	var fixedWindowType *WindowType
	func() {
		self.stateLock.Lock()
		defer self.stateLock.Unlock()
		if self.performanceProfile != nil {
			fixedWindowType = &self.performanceProfile.WindowType
		}
	}()

	if fixedWindowType != nil {
		return []WindowType{*fixedWindowType}
	} else {
		if sendPacket.ipPath.DestinationPort == 443 {
			return []WindowType{WindowTypeQuality, WindowTypeSpeed}
		}
		return []WindowType{WindowTypeSpeed, WindowTypeQuality}
	}
}

// called with stateLock
func (self *RemoteUserNatMultiClient) affinityIpPathsWithLock(ipPath *IpPath) (affinityPaths []*IpPath) {
	singleIp := false
	if self.performanceProfile != nil {
		singleIp = (self.performanceProfile.WindowSize.FixedWindowSize == 1)
	}

	if singleIp {
		singlePath := &IpPath{
			Version: ipPath.Version,
		}
		affinityPaths = append(affinityPaths, singlePath)
	} else {
		var serverNames []string
		// for 5 <= len(payload) {
		// 	tlsHeader := parseTlsHeader(payload[0:5])
		// 	if tlsHeader.valid() && 5+int(tlsHeader.contentLength) <= len(payload) {
		// 		if tlsHeader.contentType == TlsContentTypeHandshake {
		// 			// handshake
		// 			handshakeBytes := payload[5 : 5+tlsHeader.contentLength]
		// 			serverName := UnmarshalClientHelloServerName(handshakeBytes)
		// 			if serverName != "" {
		// 				serverNames = append(serverNames, serverName)
		// 			}
		// 		}
		// 	}
		// }

		// FIXME
		// if serverName == "" {
		// 	// attempt to use a reverse lookup of the associated name resolution
		// }

		if 0 < len(serverNames) {
			for _, serverName := range serverNames {
				rootDomain, err := publicsuffix.EffectiveTLDPlusOne(serverName)
				if err == nil {
					rootDomainPath := &IpPath{
						ServerName: rootDomain,
					}
					affinityPaths = append(affinityPaths, rootDomainPath)
				} else {
					serverNamePath := &IpPath{
						ServerName: serverName,
					}
					affinityPaths = append(affinityPaths, serverNamePath)
				}
			}
		} else if ipPath.DestinationPort == 80 || ipPath.DestinationPort == 53 || ipPath.DestinationPort == 443 {
			// for these ports, cycle the path per destination ip/port, regardless of protocol
			destinationPath := &IpPath{
				Version:         ipPath.Version,
				DestinationIp:   ipPath.DestinationIp,
				DestinationPort: ipPath.DestinationPort,
			}
			affinityPaths = append(affinityPaths, destinationPath)
		} else if ipPath.DestinationPort < 1024 {
			// for these ports, cycle the path per destination port, regardless of protocol or ip
			destinationPortPath := &IpPath{
				Version:         ipPath.Version,
				DestinationPort: ipPath.DestinationPort,
			}
			affinityPaths = append(affinityPaths, destinationPortPath)
		} else {
			// for user space ports, use a single path, regardless of protocol, ip, or port
			singlePath := &IpPath{
				Version: ipPath.Version,
			}
			affinityPaths = append(affinityPaths, singlePath)
		}
	}

	return
}

func (self *RemoteUserNatMultiClient) sendClientPath(ipPath *IpPath, callback func(*multiClientChannelUpdate)) {
	update, previousClient := self.sendUpdate(ipPath)
	callback(update)
	// self.updateClient(update, client)

	func() {
		self.stateLock.Lock()
		defer self.stateLock.Unlock()

		client := update.client

		if previousClient != client {
			if previousClient != nil {
				if updates, ok := self.clientUpdates[previousClient]; ok {
					delete(updates, update)
					if len(updates) == 0 {
						delete(self.clientUpdates, previousClient)
					}
				}
			}
			if client != nil && !client.IsDone() {
				updates, ok := self.clientUpdates[client]
				if !ok {
					updates = map[*multiClientChannelUpdate]bool{}
					self.clientUpdates[client] = updates
				}
				updates[update] = true
			}
		}
	}()
}

func (self *RemoteUserNatMultiClient) sendUpdate(ipPath *IpPath) (
	*multiClientChannelUpdate,
	*multiClientChannel,
) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	waitForIdle := func(update *multiClientChannelUpdate) {
		for {
			select {
			case <-update.ctx.Done():
				return
			default:
			}

			var idleTimeout time.Duration
			func() {
				self.stateLock.Lock()
				defer self.stateLock.Unlock()

				idleTimeout = update.activityTime.Add(self.settings.SequenceIdleTimeout).Sub(time.Now())
			}()
			if idleTimeout <= 0 {
				return
			} else {
				select {
				case <-update.ctx.Done():
					return
				case <-time.After(idleTimeout):
				}
			}
		}
	}

	rst := func(client *multiClientChannel) {
		if client != nil {
			// rst to destination
			if packet, ok := ipOosRst(ipPath); ok {
				client.Send(&parsedPacket{
					packet: packet,
					ipPath: ipPath,
				}, 0)
			}
		}
		// rst to source
		if packet, ok := ipOosRst(ipPath.Reverse()); ok {
			self.receivePacketCallback(TransferPath{}, protocol.ProvideMode_Network, ipPath, packet)
		}
	}

	switch ipPath.Version {
	case 4:
		ip4Path := ipPath.ToIp4Path()
		var previousClient *multiClientChannel
		update, ok := self.ip4PathUpdates[ip4Path]
		if !ok || update.IsDone() {
			update = newMultiClientChannelUpdate(self.ctx, ipPath)
			go HandleError(func() {
				defer update.cancel()

				var client *multiClientChannel
				for {
					waitForIdle(update)

					success := func() bool {
						self.stateLock.Lock()
						defer self.stateLock.Unlock()

						if t := update.activityTime.Add(self.settings.SequenceIdleTimeout).Sub(time.Now()); 0 < t {
							// updated since wait for idle
							return false
						}

						client = update.client
						update.client = nil

						delete(self.ip4PathUpdates, ip4Path)

						for affinityIp4Path, _ := range update.affinityIp4Paths {
							if paths, ok := self.affinityIp4Paths[affinityIp4Path]; ok {
								delete(paths, ip4Path)
								if len(paths) == 0 {
									delete(self.affinityIp4Paths, affinityIp4Path)
								}
							}
						}

						if client != nil {
							if updates, ok := self.clientUpdates[client]; ok {
								delete(updates, update)
								if len(updates) == 0 {
									delete(self.clientUpdates, client)
								}
							}
						}
						return true
					}()

					if success {
						break
					}
				}

				select {
				case <-self.ctx.Done():
				default:
					rst(client)
				}
			}, update.cancel)
			self.ip4PathUpdates[ip4Path] = update

			var affinityIpPaths []*IpPath
			if self.settings.DestinationAffinity {
				affinityIpPaths = self.affinityIpPathsWithLock(ipPath)
			}

			for _, affinityIpPath := range affinityIpPaths {
				affinityIp4Path := affinityIpPath.ToIp4Path()
				update.affinityIp4Paths[affinityIp4Path] = true
				paths, ok := self.affinityIp4Paths[affinityIp4Path]
				if !ok {
					paths = map[Ip4Path]time.Time{}
					self.affinityIp4Paths[affinityIp4Path] = paths
				}
				paths[ip4Path] = time.Now()

				if update.client == nil {
					var mostRecentCreateTime time.Time
					for copyIp4Path, createTime := range paths {
						if copyUpdate, ok := self.ip4PathUpdates[copyIp4Path]; ok {
							if copyUpdate.client != nil && !copyUpdate.client.IsDone() && !copyUpdate.client.isWarning() && createTime.After(mostRecentCreateTime) {
								mostRecentCreateTime = createTime
								update.client = copyUpdate.client
							}
						}
					}
				}
			}
		} else {
			previousClient = update.client
		}

		update.activityTime = time.Now()
		return update, previousClient
	case 6:
		ip6Path := ipPath.ToIp6Path()
		var previousClient *multiClientChannel
		update, ok := self.ip6PathUpdates[ip6Path]
		if !ok || update.IsDone() {
			update = newMultiClientChannelUpdate(self.ctx, ipPath)
			go HandleError(func() {
				defer update.cancel()

				var client *multiClientChannel
				for {
					waitForIdle(update)

					success := func() bool {
						self.stateLock.Lock()
						defer self.stateLock.Unlock()

						if t := update.activityTime.Add(self.settings.SequenceIdleTimeout).Sub(time.Now()); 0 < t {
							// updated since wait for idle
							return false
						}

						client = update.client
						update.client = nil

						delete(self.ip6PathUpdates, ip6Path)

						for affinityIp6Path, _ := range update.affinityIp6Paths {
							if paths, ok := self.affinityIp6Paths[affinityIp6Path]; ok {
								delete(paths, ip6Path)
								if len(paths) == 0 {
									delete(self.affinityIp6Paths, affinityIp6Path)
								}
							}
						}

						if client != nil {
							if updates, ok := self.clientUpdates[client]; ok {
								delete(updates, update)
								if len(updates) == 0 {
									delete(self.clientUpdates, client)
								}
							}
						}
						return true
					}()

					if success {
						break
					}
				}

				select {
				case <-self.ctx.Done():
				default:
					rst(client)
				}
			}, update.cancel)
			self.ip6PathUpdates[ip6Path] = update

			var affinityIpPaths []*IpPath
			if self.settings.DestinationAffinity {
				affinityIpPaths = self.affinityIpPathsWithLock(ipPath)
			}

			for _, affinityIpPath := range affinityIpPaths {
				affinityIp6Path := affinityIpPath.ToIp6Path()
				update.affinityIp6Paths[affinityIp6Path] = true
				paths, ok := self.affinityIp6Paths[affinityIp6Path]
				if !ok {
					paths = map[Ip6Path]time.Time{}
					self.affinityIp6Paths[affinityIp6Path] = paths
				}
				paths[ip6Path] = time.Now()

				if update.client == nil {
					var mostRecentCreateTime time.Time
					for copyIp6Path, createTime := range paths {
						if copyUpdate, ok := self.ip6PathUpdates[copyIp6Path]; ok {
							if copyUpdate.client != nil && !copyUpdate.client.IsDone() && !copyUpdate.client.isWarning() && createTime.After(mostRecentCreateTime) {
								mostRecentCreateTime = createTime
								update.client = copyUpdate.client
							}
						}
					}
				}
			}
		} else {
			previousClient = update.client
		}

		update.activityTime = time.Now()
		return update, previousClient
	default:
		panic(fmt.Errorf("Bad protocol version %d", ipPath.Version))
	}
}

func (self *RemoteUserNatMultiClient) receiveClientPath(ipPath *IpPath, callback func(*multiClientChannelUpdate)) bool {
	update := self.receiveUpdate(ipPath)
	if update == nil {
		return false
	}
	callback(update)
	return true
}

func (self *RemoteUserNatMultiClient) receiveUpdate(ipPath *IpPath) *multiClientChannelUpdate {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	switch ipPath.Version {
	case 4:
		ip4Path := ipPath.ToIp4Path()
		update := self.ip4PathUpdates[ip4Path]
		if update != nil {
			update.activityTime = time.Now()
			return update
		}
	case 6:
		ip6Path := ipPath.ToIp6Path()
		update := self.ip6PathUpdates[ip6Path]
		if update != nil {
			update.activityTime = time.Now()
			return update
		}
	default:
		panic(fmt.Errorf("Bad protocol version %d", ipPath.Version))
	}

	return nil
}

/*
func (self *RemoteUserNatMultiClient) updateClient(update *multiClientChannelUpdate, previousClient *multiClientChannel) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	client := update.client

	if previousClient != client {
		if previousClient != nil {
			if updates, ok := self.clientUpdates[previousClient]; ok {
				delete(updates, update)
				if len(updates) == 0 {
					delete(self.clientUpdates, previousClient)
				}
			}
		}
		if client != nil && !client.IsDone() {
			updates, ok := self.clientUpdates[client]
			if !ok {
				updates = map[*multiClientChannelUpdate]bool{}
				self.clientUpdates[client] = updates
			}
			updates[update] = true
		}
	}
}
*/

// remove a client from all updates
func (self *RemoteUserNatMultiClient) removeClient(client *multiClientChannel) {
	rstPackets := []*receivePacket{}

	func() {
		self.stateLock.Lock()
		defer self.stateLock.Unlock()

		// note client must be marked as done, otherwise it may be re-added by updates in flight
		if !client.IsDone() {
			glog.Errorf("[multi]removed client that is not marked as done. This might lead to memory leak.")
		}

		if updates, ok := self.clientUpdates[client]; ok {
			delete(self.clientUpdates, client)
			for update, _ := range updates {
				if update.client == client {
					update.client = nil

					if packet, ok := ipOosRst(update.ipPath.Reverse()); ok {
						rstPacket := &receivePacket{
							Source:      TransferPath{},
							ProvideMode: protocol.ProvideMode_Network,
							IpPath:      update.ipPath,
							Packet:      packet,
						}
						rstPackets = append(rstPackets, rstPacket)
					}
				} else {
					glog.Errorf("[multi]update associated with incorrect client")
				}
			}
		}
	}()

	select {
	case <-self.ctx.Done():
	default:
		for _, p := range rstPackets {
			self.receivePacketCallback(p.Source, p.ProvideMode, p.IpPath, p.Packet)
		}
	}
}

// `SendPacketFunction`
func (self *RemoteUserNatMultiClient) SendPacket(
	source TransferPath,
	provideMode protocol.ProvideMode,
	packet []byte,
	timeout time.Duration,
) bool {
	minRelationship := max(provideMode, self.provideMode)

	ipPath, payload, err := ParseIpPathWithPayload(packet)
	if err != nil {
		glog.Infof("[multi]send bad packet = %s\n", err)
		return false
	}
	r, err := self.securityPolicy.Inspect(minRelationship, ipPath)
	if err != nil {
		glog.Infof("[multi]send bad packet = %s\n", err)
		return false
	}
	switch r {
	case SecurityPolicyResultAllow:
		parsedPacket := &parsedPacket{
			packet:  packet,
			ipPath:  ipPath,
			payload: payload,
		}
		return self.sendPacket(source, provideMode, parsedPacket, timeout)
	case SecurityPolicyResultDrop:
		if self.LocalSecurityBypass() {
			return self.localUserNat.SendPacket(source, provideMode, packet, timeout)
		} else {
			return false
		}
	default:
		glog.V(1).Infof("[multi]drop packet ipv%d p%v -> %s:%d\n", ipPath.Version, ipPath.Protocol, ipPath.DestinationIp, ipPath.DestinationPort)
		return false
	}
}

func (self *RemoteUserNatMultiClient) canSendPacket(sendPacket *parsedPacket, update *multiClientChannelUpdate) (allow bool) {
	ipPath := sendPacket.ipPath
	switch ipPath.Protocol {
	case IpProtocolTcp:
		if self.settings.TcpCollapsePrevention {
			// limit sender tcp collapse
			// as soon as a packet is sent to a client, either the client will eith reliabily transfer the packet,
			// or the client will be dropped
			// retransmits don't need to be sent as soon as the packet is committed to a client
			if ipPath.Syn || ipPath.Rst {
				allow = true
			} else {
				func() {
					self.stateLock.Lock()
					defer self.stateLock.Unlock()

					if update.canUpdateSequence(sendPacket) {
						allow = true
						return
					}
				}()
			}
		} else {
			allow = true
		}
	default:
		allow = true
	}
	return
}

func (self *RemoteUserNatMultiClient) sendPacket(
	source TransferPath,
	provideMode protocol.ProvideMode,
	sendPacket *parsedPacket,
	timeout time.Duration,
) (success bool) {
	ipPath := sendPacket.ipPath
	self.sendClientPath(ipPath, func(update *multiClientChannelUpdate) {
		if !self.canSendPacket(sendPacket, update) {
			return
		}

		enterTime := time.Now()

		if ipPath.Syn || ipPath.Rst {
			func() {
				self.stateLock.Lock()
				defer self.stateLock.Unlock()
				update.resetSequence(sendPacket)
			}()
		}

		currentClient := func() *multiClientChannel {
			self.stateLock.Lock()
			defer self.stateLock.Unlock()
			return update.client
		}
		for client := currentClient(); client != nil; {
			var err error
			success, err = client.SendDetailed(sendPacket, timeout)
			if success {
				func() {
					self.stateLock.Lock()
					defer self.stateLock.Unlock()
					update.updateSequence(sendPacket)
				}()
			} else if err != nil {
				// reset the path

				glog.Infof("[multi]reset error = %s\n", err)

				func() {
					self.stateLock.Lock()
					defer self.stateLock.Unlock()

					update.client = nil
				}()

				rstPackets := []*receivePacket{}

				if packet, ok := ipOosRst(update.ipPath.Reverse()); ok {
					rstPacket := &receivePacket{
						Source:      TransferPath{},
						ProvideMode: protocol.ProvideMode_Network,
						IpPath:      update.ipPath,
						Packet:      packet,
					}
					rstPackets = append(rstPackets, rstPacket)
				}

				select {
				case <-self.ctx.Done():
				default:
					for _, p := range rstPackets {
						self.receivePacketCallback(p.Source, p.ProvideMode, p.IpPath, p.Packet)
					}
				}
			}
			// else the packet was dropped due to backpressure
			// keep sending to the client until there is an error
			return
		}

		// find a new client
		// the race is between as many clients as can send in parallel

		// if _, fixed := self.generator.FixedDestinationSize(); fixed {
		// 	window := self.windows[WindowTypeQuality]
		// 	orderedClients := window.OrderedClients()

		// 	for _, client := range orderedClients {
		// 		if client.Send(sendPacket, timeout) {
		// 			success = true

		// 			func() {
		// 				self.stateLock.Lock()
		// 				defer self.stateLock.Unlock()

		// 				update.client = client
		// 			}()
		// 		}
		// 	}

		// 	return
		// }

		raceClients := func(orderedClients []*multiClientChannel, sendTimeout time.Duration) {
			switch len(orderedClients) {
			case 0:
				return
			case 1:
				// send to one client, no race
				client := orderedClients[0]
				if client.Send(sendPacket, sendTimeout) {
					success = true

					func() {
						self.stateLock.Lock()
						defer self.stateLock.Unlock()

						update.client = client
					}()
				}
				return

			default:

				defer func() {
					if success {
						MessagePoolReturn(sendPacket.packet)
					}
				}()

				var successCount atomic.Int32

				send := func(client *multiClientChannel) {
					select {
					case <-update.ctx.Done():
						return
					default:
					}

					done := false

					func() {
						self.stateLock.Lock()
						defer self.stateLock.Unlock()

						if update.client != nil {
							// another client already chosen, done
							done = true
							return
						}
					}()
					if done {
						return
					}

					p := &parsedPacket{
						packet: MessagePoolShareReadOnly(sendPacket.packet),
						ipPath: update.ipPath,
					}
					if client.SendWithAck(p, sendTimeout, true) {
						successCount.Add(1)

						var initRace *multiClientChannelUpdateRace
						var initRaceEarlyComplete <-chan struct{}
						var abandonedClients []*multiClientChannel
						func() {
							self.stateLock.Lock()
							defer self.stateLock.Unlock()

							if update.client != nil {
								// another client already chosen, done
								done = true
								return
							}

							race := update.race
							if race == nil {
								update.initRace()
								race = update.race

								initRace = race
								initRaceEarlyComplete = race.completeMonitor.NotifyChannel()
							}
							state := race.clientStates[client]
							if state == nil {
								state = &multiClientChannelRaceClientState{
									sendTime: time.Now(),
								}
								race.clientStates[client] = state
							}
							state.sentPacketCount += 1
							race.sentPacketCount += 1
							bufferExceeded := state != nil && self.settings.MultiRaceSetOnNoResponseTimeout <= time.Now().Sub(state.sendTime) || self.settings.MultiRaceClientSentPacketMaxCount < state.sentPacketCount
							if race.packetCount == 0 && bufferExceeded {
								// no client response in timeout, lock in this client
								// this happens for example when the client only sends and does not receive (e.g. udp send)

								for abandonedClient, _ := range race.clientStates {
									if abandonedClient != client {
										abandonedClients = append(abandonedClients, abandonedClient)
									}
								}

								update.clearRace()
								update.client = client
							}
						}()

						if initRace != nil {
							self.scheduleCompleteRace(update.ipPath, initRace, initRaceEarlyComplete)
						}

						if 0 < len(abandonedClients) {
							if rstPacket, ok := ipOosRst(update.ipPath); ok {
								for _, abandonedClient := range abandonedClients {
									abandonedClient.Send(&parsedPacket{
										packet: rstPacket,
										ipPath: update.ipPath,
									}, 0)
								}
							}
						}
					} else {
						MessagePoolReturn(p.packet)
					}
				}

				var raceOrderedClients []*multiClientChannel
				if 0 < self.settings.MultiRaceClientCount && self.settings.MultiRaceClientCount < len(orderedClients) {
					raceOrderedClients = orderedClients[:self.settings.MultiRaceClientCount]
				} else {
					raceOrderedClients = orderedClients
				}

				// if 0 < timeout {
				var wg sync.WaitGroup

				for _, client := range raceOrderedClients {
					wg.Add(1)
					go HandleError(func() {
						defer wg.Done()

						send(client)
					})
				}

				wg.Wait()
				// } else {
				// 	for _, client := range raceOrderedClients {
				// 		send(client)
				// 	}
				// }

				if 0 < successCount.Load() {
					success = true
				}
				return
			}
		}

		coalesceOrderedClients := func() []*multiClientChannel {
			for _, windowType := range self.selectWindowTypes(sendPacket) {
				if window, ok := self.windows[windowType]; ok {
					orderedClients := window.OrderedClients()
					if 0 < len(orderedClients) {
						return orderedClients
					}
				}
			}
			return []*multiClientChannel{}
		}

		raceClients(coalesceOrderedClients(), 0)
		if success {
			return
		}

		for {
			var retryTimeout time.Duration
			if 0 <= timeout {
				remainingTimeout := enterTime.Add(timeout).Sub(time.Now())

				if remainingTimeout <= 0 {
					// drop
					return
				}

				retryTimeout = min(remainingTimeout, self.settings.SendRetryTimeout)
			} else {
				retryTimeout = self.settings.SendRetryTimeout
			}

			startTime := time.Now()
			raceClients(coalesceOrderedClients(), retryTimeout)
			if success {
				return
			}
			endTime := time.Now()
			retryTimeout -= endTime.Sub(startTime)

			if 0 < retryTimeout {
				select {
				case <-update.ctx.Done():
					return
				case <-time.After(retryTimeout):
				}
			}
		}
	})
	return
}

// clientReceivePacketFunction
func (self *RemoteUserNatMultiClient) clientReceivePacket(
	sourceClient *multiClientChannel,
	source TransferPath,
	provideMode protocol.ProvideMode,
	ipPath *IpPath,
	packet []byte,
) {
	r, err := self.ingressSecurityPolicy.Inspect(provideMode, ipPath)
	if err != nil {
		return
	}
	if r != SecurityPolicyResultAllow {
		return
	}

	ipPath = ipPath.Reverse()

	var abandonedClients []*multiClientChannel
	var receivePackets []*receivePacket
	var returnPackets []*receivePacket
	success := self.receiveClientPath(ipPath, func(update *multiClientChannelUpdate) {
		self.stateLock.Lock()
		defer self.stateLock.Unlock()

		client := update.client

		if client == sourceClient {
			p := &receivePacket{
				Source:      source,
				ProvideMode: provideMode,
				IpPath:      ipPath,
				Packet:      packet,
			}
			receivePackets = []*receivePacket{p}
		} else if client != nil {
			// another client already chosen, drop
		} else if race := update.race; race == nil {
			// no race, no client, drop
			glog.Infof("[multi]receive no race and no client")
		} else if state, ok := race.clientStates[sourceClient]; !ok {
			// this client is not part of the race, drop
			glog.Infof("[multi]receive client not part of race")
		} else if len(state.packets) < self.settings.MultiRaceClientPacketMaxCount && race.packetCount < self.settings.MultiRacePacketMaxCount {
			packetCopy, pooled := MessagePoolCopyDetailed(packet)
			receivePacket := &receivePacket{
				Source:      source,
				ProvideMode: provideMode,
				IpPath:      ipPath,
				Packet:      packetCopy,
				Pooled:      pooled,
			}
			state.packets = append(state.packets, receivePacket)
			if 1 == len(state.packets) {
				state.receiveTime = time.Now()
			}
			race.packetCount += 1
			/*
				if race.packetCount == 1 {
					// schedule the race evaluation on first packet
					earlyComplete := race.completeMonitor.NotifyChannel()
					// copy the ip path since the first packet may not be ultimately retained to the end of the race
					self.scheduleCompleteRace(ipPathCopy, race, earlyComplete)
				}
			*/
			if len(state.packets) == 1 {
				race.clientsWithPacketCount += 1
				if int(float32(len(race.clientStates))*self.settings.MultiRaceClientEarlyCompleteFraction) <= race.clientsWithPacketCount {
					race.completeMonitor.NotifyAll()
				}
			}
		} else {
			// race buffer limits exceeded, end the race immediately
			glog.Infof("[multi]receive race buffer limit reached")

			for abandonedClient, abandonedState := range race.clientStates {
				if abandonedClient != client {
					abandonedClients = append(abandonedClients, abandonedClient)
					for _, p := range abandonedState.packets {
						if p.Pooled {
							p.Pooled = false
							returnPackets = append(returnPackets, p)
						}
					}
				}
			}

			update.clearRace()
			update.client = client
			receivePacket := &receivePacket{
				Source:      source,
				ProvideMode: provideMode,
				IpPath:      ipPath,
				Packet:      packet,
			}
			receivePackets = append(state.packets, receivePacket)
			for _, p := range receivePackets {
				if p.Pooled {
					p.Pooled = false
					returnPackets = append(returnPackets, p)
				}
			}
		}
	})
	if success {
		if 0 < len(abandonedClients) {
			if rstPacket, ok := ipOosRst(ipPath); ok {
				for _, abandonedClient := range abandonedClients {
					abandonedClient.Send(&parsedPacket{
						packet: rstPacket,
						ipPath: ipPath,
					}, 0)
				}
			}
		}
		for _, p := range receivePackets {
			self.receivePacketCallback(p.Source, p.ProvideMode, p.IpPath, p.Packet)
		}
		for _, p := range returnPackets {
			MessagePoolReturn(p.Packet)
		}
	} else {
		// incoming packets not in response to outgoing packets
		self.receivePacketCallback(source, provideMode, ipPath, packet)
	}
}

// this can be called inside the state lock
func (self *RemoteUserNatMultiClient) scheduleCompleteRace(
	ipPath *IpPath,
	race *multiClientChannelUpdateRace,
	earlyComplete <-chan struct{},
) {
	go HandleError(func() {
		// wait for the race to finish, then choose

		select {
		case <-race.ctx.Done():
			return
		case <-earlyComplete:
		case <-time.After(self.settings.MultiRaceSetOnResponseTimeout):
		}

		var abandonedClients []*multiClientChannel
		var receivePackets []*receivePacket
		var returnPackets []*receivePacket
		self.receiveClientPath(ipPath, func(update *multiClientChannelUpdate) {
			self.stateLock.Lock()
			defer self.stateLock.Unlock()

			if update.race == race {
				defer update.clearRace()

				if update.client == nil {

					// weighted shuffle clients by rtt
					orderedClients := []*multiClientChannel{}
					weights := map[*multiClientChannel]float32{}
					for client, state := range race.clientStates {
						if 0 < len(state.packets) {
							orderedClients = append(orderedClients, client)
							rtt := state.receiveTime.Sub(state.sendTime)
							weights[client] = float32(rtt / time.Millisecond)
						}
					}
					WeightedShuffleWithEntropy(orderedClients, weights, self.settings.StatsWindowEntropy)

					if 0 < len(orderedClients) {
						// the last is the lowest rtt
						client := orderedClients[len(orderedClients)-1]

						update.client = client
						receivePackets = race.clientStates[client].packets
						for _, p := range receivePackets {
							if p.Pooled {
								returnPackets = append(returnPackets, p)
							}
						}
					}
				}
				// else the client is already set
				for abandonedClient, abandonedState := range race.clientStates {
					if abandonedClient != update.client {
						abandonedClients = append(abandonedClients, abandonedClient)
						for _, p := range abandonedState.packets {
							if p.Pooled {
								returnPackets = append(returnPackets, p)
							}
						}
					}
				}
			}
			// else the client is on a new race
		})
		if 0 < len(abandonedClients) {
			if rstPacket, ok := ipOosRst(ipPath); ok {
				for _, abandonedClient := range abandonedClients {
					abandonedClient.Send(&parsedPacket{
						packet: rstPacket,
						ipPath: ipPath,
					}, 0)
				}
			}
		}
		for _, p := range receivePackets {
			self.receivePacketCallback(p.Source, p.ProvideMode, p.IpPath, p.Packet)
		}
		for _, p := range returnPackets {
			MessagePoolReturn(p.Packet)
		}
	})
}

func (self *RemoteUserNatMultiClient) Shuffle() {
	for _, window := range self.windows {
		window.shuffle()
	}
}

func (self *RemoteUserNatMultiClient) Close() {
	self.cancel()
	for _, window := range self.windows {
		window.Close()
	}

	func() {
		self.stateLock.Lock()
		defer self.stateLock.Unlock()

		removedUpdates := []*multiClientChannelUpdate{}

		for _, update := range self.ip4PathUpdates {
			removedUpdates = append(removedUpdates, update)
			// update.Close()
		}
		for _, update := range self.ip6PathUpdates {
			removedUpdates = append(removedUpdates, update)
			// update.Close()
		}
		clear(self.ip4PathUpdates)
		clear(self.ip6PathUpdates)
		clear(self.affinityIp4Paths)
		clear(self.affinityIp6Paths)
		// clear(self.updateIp4Paths)
		// clear(self.updateIp6Paths)
		// clear(clientUpdates)

		for _, update := range removedUpdates {
			update.Close()
		}
	}()

	self.localUserNat.Close()
	self.localUserNatUnsub()
}

type multiClientChannelUpdate struct {
	ctx    context.Context
	cancel context.CancelFunc

	client       *multiClientChannel
	race         *multiClientChannelUpdateRace
	activityTime time.Time
	ipPath       *IpPath

	sequencePacketCount int
	ackSequenceNumber   uint32
	sequenceNumber      uint32
	sequenceNumber64    uint64

	affinityIp4Paths map[Ip4Path]bool
	affinityIp6Paths map[Ip6Path]bool
}

func newMultiClientChannelUpdate(ctx context.Context, ipPath *IpPath) *multiClientChannelUpdate {
	cancelCtx, cancel := context.WithCancel(ctx)
	return &multiClientChannelUpdate{
		ctx:              cancelCtx,
		cancel:           cancel,
		ipPath:           ipPath,
		affinityIp4Paths: map[Ip4Path]bool{},
		affinityIp6Paths: map[Ip6Path]bool{},
	}
}

func (self *multiClientChannelUpdate) resetSequence(sendPacket *parsedPacket) {
	ipPath := sendPacket.ipPath

	nextAckSequenceNumber := ipPath.AckSequenceNumber
	self.ackSequenceNumber = nextAckSequenceNumber

	nextSequenceNumber64 := uint64(ipPath.SequenceNumber)
	nextSequenceNumber := uint32(nextSequenceNumber64)
	self.sequenceNumber = nextSequenceNumber
	self.sequenceNumber64 = nextSequenceNumber64

	self.sequencePacketCount = 0
}

func (self *multiClientChannelUpdate) updateSequence(sendPacket *parsedPacket) {
	ipPath := sendPacket.ipPath
	update := false

	nextAckSequenceNumber := ipPath.AckSequenceNumber
	if self.ackSequenceNumber != nextAckSequenceNumber {
		self.ackSequenceNumber = nextAckSequenceNumber
		update = true
	}

	nextSequenceNumber64 := uint64(ipPath.SequenceNumber) + uint64(len(sendPacket.payload))
	nextSequenceNumber := uint32(nextSequenceNumber64)
	if self.sequenceNumber < nextSequenceNumber || self.sequenceNumber64 < nextSequenceNumber64 {
		self.sequenceNumber = nextSequenceNumber
		self.sequenceNumber64 = nextSequenceNumber64
		update = true
	}

	if update {
		self.sequencePacketCount += 1
	}
}

func (self *multiClientChannelUpdate) canUpdateSequence(sendPacket *parsedPacket) bool {
	if self.sequencePacketCount == 0 {
		return true
	}

	ipPath := sendPacket.ipPath

	if self.ackSequenceNumber != ipPath.AckSequenceNumber {
		return true
	}

	nextSequenceNumber64 := uint64(ipPath.SequenceNumber) + uint64(len(sendPacket.payload))
	nextSequenceNumber := uint32(nextSequenceNumber64)
	if self.sequenceNumber <= nextSequenceNumber || self.sequenceNumber64 <= nextSequenceNumber64 {
		return true
	}

	return false
}

func (self *multiClientChannelUpdate) initRace() {
	if self.race == nil {
		self.race = newMultiClientChannelUpdateRace(self.ctx)
	}
}

func (self *multiClientChannelUpdate) clearRace() {
	if self.race != nil {
		self.race.Close()
		self.race = nil
	}
}

func (self *multiClientChannelUpdate) IsDone() bool {
	select {
	case <-self.ctx.Done():
		return true
	default:
		return false
	}
}

func (self *multiClientChannelUpdate) Close() {
	self.cancel()
	// self.client = nil
	self.clearRace()
}

type multiClientChannelUpdateRace struct {
	ctx                    context.Context
	cancel                 context.CancelFunc
	clientStates           map[*multiClientChannel]*multiClientChannelRaceClientState
	sentPacketCount        int
	packetCount            int
	clientsWithPacketCount int
	completeMonitor        *Monitor
}

func newMultiClientChannelUpdateRace(ctx context.Context) *multiClientChannelUpdateRace {
	cancelCtx, cancel := context.WithCancel(ctx)
	return &multiClientChannelUpdateRace{
		ctx:          cancelCtx,
		cancel:       cancel,
		clientStates: map[*multiClientChannel]*multiClientChannelRaceClientState{},
		// sentPacketCount:        0,
		// packetCount:            0,
		// clientsWithPacketCount: 0,
		completeMonitor: NewMonitor(),
	}
}

func (self *multiClientChannelUpdateRace) Close() {
	self.cancel()
	// for _, state := range self.clientStates {
	// 	state.packets = nil
	// }
	// clear(self.clientStates)
}

type multiClientChannelRaceClientState struct {
	sendTime        time.Time
	receiveTime     time.Time
	packets         []*receivePacket
	sentPacketCount int
}

type parsedPacket struct {
	packet  []byte
	ipPath  *IpPath
	payload []byte
}

func newParsedPacket(packet []byte) (*parsedPacket, error) {
	ipPath, err := ParseIpPath(packet)
	if err != nil {
		return nil, err
	}
	return &parsedPacket{
		packet: packet,
		ipPath: ipPath,
	}, nil
}

type multiClientWindow struct {
	ctx    context.Context
	cancel context.CancelFunc

	generator                   MultiClientGenerator
	clientReceivePacketCallback clientReceivePacketFunction
	ingressSecurityPolicy       SecurityPolicy
	clientRemoveCallback        func(client *multiClientChannel)
	windowType                  WindowType

	settings *MultiClientSettings

	clientChannelArgs chan *multiClientChannelArgs

	monitor *RemoteUserNatMultiClientMonitor

	contractStatusCallbacks *CallbackList[ContractStatusFunction]

	stateLock          sync.Mutex
	clients            map[Id]*multiClientChannel
	performanceProfile *PerformanceProfile

	generatorMonitor *Monitor
	resizeMonitor    *Monitor
}

func newMultiClientWindow(
	ctx context.Context,
	cancel context.CancelFunc,
	generator MultiClientGenerator,
	clientReceivePacketCallback clientReceivePacketFunction,
	ingressSecurityPolicy SecurityPolicy,
	clientRemoveCallback func(client *multiClientChannel),
	windowType WindowType,
	settings *MultiClientSettings,
) *multiClientWindow {
	window := &multiClientWindow{
		ctx:                         ctx,
		cancel:                      cancel,
		generator:                   generator,
		clientReceivePacketCallback: clientReceivePacketCallback,
		ingressSecurityPolicy:       ingressSecurityPolicy,
		clientRemoveCallback:        clientRemoveCallback,
		windowType:                  windowType,
		settings:                    settings,
		clientChannelArgs:           make(chan *multiClientChannelArgs),
		monitor:                     NewRemoteUserNatMultiClientMonitor(&settings.RemoteUserNatMultiClientMonitorSettings),
		contractStatusCallbacks:     NewCallbackList[ContractStatusFunction](),
		clients:                     map[Id]*multiClientChannel{},
		generatorMonitor:            NewMonitor(),
		resizeMonitor:               NewMonitor(),
	}

	go HandleError(window.randomEnumerateClientArgs, cancel)
	go HandleError(window.resize, cancel)

	return window
}

func (self *multiClientWindow) AddContractStatusCallback(contractStatusCallback ContractStatusFunction) func() {
	callbackId := self.contractStatusCallbacks.Add(contractStatusCallback)
	return func() {
		self.contractStatusCallbacks.Remove(callbackId)
	}
}

func (self *multiClientWindow) contractStatus(contractStatus *ContractStatus) {
	for _, contractStatusCallback := range self.contractStatusCallbacks.Get() {
		HandleError(func() {
			contractStatusCallback(contractStatus)
		})
	}
}

// the performance profile will take effect at the next `resize` iteration
func (self *multiClientWindow) SetPerformanceProfile(performanceProfile *PerformanceProfile) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	self.performanceProfile = performanceProfile
}

func (self *multiClientWindow) randomEnumerateClientArgs() {
	defer func() {
		close(self.clientChannelArgs)

		// drain the channel
		func() {
			for {
				select {
				case args, ok := <-self.clientChannelArgs:
					if !ok {
						return
					}
					self.generator.RemoveClientArgs(&args.MultiClientGeneratorClientArgs)
				}
			}
		}()
	}()

	for {
		generatorNotify := self.generatorMonitor.NotifyChannel()

		// exclude healthy destinations that are already in the window
		windowDestinations := func() map[MultiHopId]bool {
			self.stateLock.Lock()
			defer self.stateLock.Unlock()
			windowDestinations := map[MultiHopId]bool{}
			for _, client := range self.clients {
				if !client.isWarning() {
					windowDestinations[client.Destination()] = true
				}
			}
			return windowDestinations
		}

		destinations, err := self.generator.NextDestinations(
			self.settings.WindowExpandBlockCount,
			maps.Keys(windowDestinations()),
			self.windowType.RankMode(),
		)
		if err != nil {
			glog.Infof("[multi]window enumerate error timeout = %s\n", err)
			select {
			case <-self.ctx.Done():
				return
			case <-time.After(self.settings.WindowEnumerateErrorTimeout):
			}
			continue
		}

		func() {
			// destinations must be used by `expirationTime`
			expirationTime := time.Now().Add(self.settings.WindowExpandArgsTimeout)
			for destination, stats := range destinations {

				for {
					timeout := expirationTime.Sub(time.Now())
					if timeout <= 0 {
						return
					}

					clientArgs, err := self.generator.NewClientArgs()
					if err != nil {
						glog.Infof("[multi]create client args error = %s\n", err)
						select {
						case <-self.ctx.Done():
							return
						case <-time.After(self.settings.WindowEnumerateErrorTimeout):
						}
						continue
					}

					args := &multiClientChannelArgs{
						Destination:                    destination,
						DestinationStats:               stats,
						MultiClientGeneratorClientArgs: *clientArgs,
					}
					select {
					case <-self.ctx.Done():
						self.generator.RemoveClientArgs(clientArgs)
						return
					case self.clientChannelArgs <- args:
					case <-time.After(timeout):
						// destination expired
						glog.Infof("[multi]create client args expired\n")
						self.generator.RemoveClientArgs(clientArgs)
						return
					}
					break
				}
			}
		}()

		select {
		case <-self.ctx.Done():
			return
		case <-generatorNotify:
		}
	}
}

func (self *multiClientWindow) resize() {
	for {
		update := self.resizeMonitor.NotifyChannel()

		var windowSize WindowSizeSettings
		var fixedWindowType *WindowType
		func() {
			self.stateLock.Lock()
			defer self.stateLock.Unlock()
			if self.performanceProfile != nil {
				fixedWindowType = &self.performanceProfile.WindowType
				windowSize = self.performanceProfile.WindowSize
			} else {
				windowSize = self.settings.WindowSizes[self.windowType]
			}
		}()

		startTime := time.Now()

		warnedClients := []*multiClientChannel{}
		clients := []*multiClientChannel{}
		maxSourceCount := 0
		weights := map[*multiClientChannel]float32{}
		durations := map[*multiClientChannel]time.Duration{}

		removeClient := func(client *multiClientChannel) {
			client.Close()
			func() {
				self.stateLock.Lock()
				defer self.stateLock.Unlock()
				delete(self.clients, client.ClientId())
			}()
			self.removeClients(client)
		}
		keepClient := func(client *multiClientChannel, stats *clientWindowStats) {
			clients = append(clients, client)
			maxSourceCount = max(maxSourceCount, stats.sourceCount)
			weights[client] = float32(stats.EffectiveByteCountPerSecond())
			durations[client] = stats.clientDuration
		}
		warnClient := func(client *multiClientChannel, stats *clientWindowStats) {
			warnedClients = append(warnedClients, client)
			maxSourceCount = max(maxSourceCount, stats.sourceCount)
			weights[client] = float32(stats.EffectiveByteCountPerSecond())
			durations[client] = stats.clientDuration
		}

		clientStats := map[*multiClientChannel]*clientWindowStats{}
		for _, client := range self.unorderedClients() {
			if stats, err := client.WindowStats(); err == nil {
				clientStats[client] = stats
			} else {
				glog.Infof("[multi]remove error client [%s] = %s\n", client.ClientId(), err)
				removeClient(client)
			}
		}

		netHealthRanks := map[*multiClientChannel]int{}
		func() {
			orderedClients := maps.Keys(clientStats)
			slices.SortFunc(orderedClients, func(a *multiClientChannel, b *multiClientChannel) int {
				// descending healthy duration, net healthy duration
				statsA := clientStats[a]
				statsB := clientStats[b]

				if c := statsB.healthyDuration - statsA.healthyDuration; c < 0 {
					return -1
				} else if 0 < c {
					return 1
				}

				if c := statsB.netHealthyDuration - statsA.netHealthyDuration; c < 0 {
					return -1
				} else if 0 < c {
					return 1
				}

				return 0
			})
			for i, client := range orderedClients {
				netHealthRanks[client] = i
			}
		}()

		for client, stats := range clientStats {
			// note for fixed destination size, the destination might still be aliased with multiple clients
			// TODO it's still not clear why one client might stop working occasionally

			healthy := stats.unhealthyDuration < self.settings.StatsWindowMaxUnhealthyDuration

			printStats := func(status string) {
				effectiveByteCountPerSecond := stats.EffectiveByteCountPerSecond()
				expectedByteCountPerSecond := stats.ExpectedByteCountPerSecond()

				glog.V(1).Infof(
					"[multi]%s [%s]: h=%d+%dms/u=%d+%dms effective=%db/s expected=%db/s send=%db sendNack=%db receive=%db\n",
					status,
					client.ClientId(),
					stats.netHealthyDuration/time.Millisecond,
					stats.healthyDuration/time.Millisecond,
					stats.netUnhealthyDuration/time.Millisecond,
					stats.unhealthyDuration/time.Millisecond,
					effectiveByteCountPerSecond,
					expectedByteCountPerSecond,
					stats.sendAckByteCount,
					stats.sendNackByteCount,
					stats.receiveAckByteCount,
				)
			}
			// the top `StatsWindowKeepHealthiestCount` won't be marked as warning or removed
			netHealthRank := netHealthRanks[client]
			remove := max(windowSize.FixedWindowSize, windowSize.KeepHealthiestCount) <= netHealthRank
			if healthy {
				// a client after its `removeTime` will be in a permananent warning state as long as it continues to route traffic
				// this prevents new connections from using the client
				if stats.unhealthyDuration < self.settings.StatsWindowWarnUnhealthyDuration {
					if !stats.removeTime.IsZero() && stats.removeTime.Before(startTime) {
						printStats("client drain")
						client.setWarning(remove)
						warnClient(client, stats)
					} else {
						printStats("client ok")
						windowSizeUlimit := self.settings.DefaultUlimit
						if 0 < windowSize.Ulimit {
							windowSizeUlimit = windowSize.Ulimit
						}
						ulimit := 0 < windowSizeUlimit && windowSizeUlimit <= stats.netSourceCount
						if ulimit {
							client.setWarning(true)
							warnClient(client, stats)
						} else {
							client.setWarning(false)
							keepClient(client, stats)
						}
					}
				} else {
					printStats("client health warning")
					client.setWarning(remove)
					warnClient(client, stats)
				}
			} else {
				printStats(fmt.Sprintf("unhealthy client (#%d remove=%t)", netHealthRank, remove))

				if remove {
					client.setWarning(true)
					removeClient(client)
				} else {
					client.setWarning(false)
					warnClient(client, stats)
				}
			}
		}

		collapseLowestWeighted := func(targetWindowSize int) {

			n := (len(warnedClients) + len(clients)) - targetWindowSize

			collapse := func(cs []*multiClientChannel) {
				if 0 < n && 0 < len(cs) {
					m := min(len(cs), n)
					if 0 < m {
						slices.SortFunc(cs, func(a *multiClientChannel, b *multiClientChannel) int {
							// descending weight
							aWeight := weights[a]
							bWeight := weights[b]
							if aWeight < bWeight {
								return 1
							} else if bWeight < aWeight {
								return -1
							}
							aDuration := durations[a]
							bDuration := durations[b]
							if aDuration < bDuration {
								return 1
							} else if bDuration < aDuration {
								return -1
							}
							return 0
						})

						for _, client := range cs[len(cs)-m : len(cs)] {
							if self.settings.StatsWindowGraceperiod <= durations[client] && weights[client] <= 0 {
								removeClient(client)
							}
						}
						n -= m
					}
				}
			}
			collapse(warnedClients)
			collapse(clients)
		}

		p2pOnlyWindowSize := 0
		for _, client := range clients {
			if client.IsP2pOnly() {
				p2pOnlyWindowSize += 1
			}
		}

		var windowSizeMin int
		var targetWindowSize int
		if fixedDestinationSize, fixed := self.generator.FixedDestinationSize(); fixed {
			targetWindowSize = fixedDestinationSize
			windowSizeMin = targetWindowSize
		} else if 0 < windowSize.FixedWindowSize {
			if fixedWindowType == nil || self.windowType == *fixedWindowType {
				targetWindowSize = windowSize.FixedWindowSize
			} else {
				// not the active window, disable resize
				targetWindowSize = 0
			}
			windowSizeMin = targetWindowSize
		} else {
			// scale the number of reconnects
			reconnectScale := self.settings.DefaultReconnectScale
			if 0 < windowSize.WindowSizeReconnectScale {
				reconnectScale = windowSize.WindowSizeReconnectScale
			}
			targetWindowSize = int(math.Ceil(float64(maxSourceCount) * reconnectScale))

			if n := windowSize.WindowSizeMinP2pOnly - p2pOnlyWindowSize; 0 < n {
				targetWindowSize += n
			}

			targetWindowSize = min(
				windowSize.WindowSizeMax,
				max(
					windowSize.WindowSizeMin,
					targetWindowSize,
				),
			)
			windowSizeMin = windowSize.WindowSizeMin
		}

		addedCount := 0
		if len(clients) < targetWindowSize {
			// expand
			n := targetWindowSize - len(clients)
			self.monitor.AddWindowExpandEvent(
				windowSizeMin <= len(clients),
				targetWindowSize+len(warnedClients),
			)
			addedCount = self.expand(
				windowSize,
				len(clients),
				p2pOnlyWindowSize,
				targetWindowSize,
				windowSizeMin,
				targetWindowSize-len(clients),
			)
			glog.V(1).Infof("[multi]window expand +%d %d->%d (+%d)\n", n, len(clients), targetWindowSize, addedCount)
		}
		if 0 < windowSize.WindowSizeHardMax && windowSize.WindowSizeHardMax < len(clients)+len(warnedClients)+addedCount {
			self.monitor.AddWindowExpandEvent(
				windowSizeMin <= len(clients)+addedCount,
				windowSize.WindowSizeHardMax,
			)
			collapseLowestWeighted(max(0, windowSize.WindowSizeHardMax-addedCount))
			glog.V(1).Infof("[multi]window collapse -%d ->%d\n", (len(clients)+len(warnedClients)+addedCount)-windowSize.WindowSizeHardMax, windowSize.WindowSizeHardMax)
		} else {
			self.monitor.AddWindowExpandEvent(
				windowSizeMin <= len(clients)+addedCount,
				len(clients)+len(warnedClients)+addedCount,
			)
		}

		timeout := self.settings.WindowResizeTimeout - time.Now().Sub(startTime)
		if timeout <= 0 {
			select {
			case <-self.ctx.Done():
				return
			default:
			}
		} else {
			select {
			case <-self.ctx.Done():
				return
			case <-update:
			case <-time.After(timeout):
			}
		}
	}
}

func (self *multiClientWindow) expand(
	windowSize WindowSizeSettings,
	currentWindowSize int,
	currentP2pOnlyWindowSize int,
	targetWindowSize int,
	windowSizeMin int,
	n int,
) (returnPingSuccess int) {
	mutex := sync.Mutex{}
	pendingPingDones := []context.Context{}
	added := 0
	addedP2pOnly := 0
	pingSuccess := 0

	defer func() {
		mutex.Lock()
		defer mutex.Unlock()

		returnPingSuccess = pingSuccess
	}()

	endTime := time.Now().Add(self.settings.WindowExpandTimeout)

	for i := 0; i < n; i += 1 {
		timeout := endTime.Sub(time.Now())
		if timeout < 0 {
			glog.V(1).Infof("[multi]expand window timeout\n")
			return
		}

		self.generatorMonitor.NotifyAll()
		select {
		case <-self.ctx.Done():
			return
		// case <- update:
		//     // continue
		case args, ok := <-self.clientChannelArgs:
			if !ok {
				return
			}
			// func() {
			// 	self.stateLock.Lock()
			// 	defer self.stateLock.Unlock()
			// 	_, ok = self.destinationClients[args.Destination]
			// }()

			// if ok {
			// 	// already have a client in the window for this destination
			// 	self.generator.RemoveClientArgs(&args.MultiClientGeneratorClientArgs)
			// } else {
			// randomly set to p2p only to meet the minimum requirement
			if !args.MultiClientGeneratorClientArgs.P2pOnly {
				var a int
				var b int
				func() {
					mutex.Lock()
					defer mutex.Unlock()

					a = max(windowSize.WindowSizeMin-(currentWindowSize+added), 0)
					b = max(windowSize.WindowSizeMinP2pOnly-(currentP2pOnlyWindowSize+addedP2pOnly), 0)
				}()
				var p2pOnlyP float32
				if a+b == 0 {
					p2pOnlyP = 0
				} else {
					p2pOnlyP = float32(b) / float32(a+b)
				}
				args.MultiClientGeneratorClientArgs.P2pOnly = mathrand.Float32() < p2pOnlyP
			}

			client, err := newMultiClientChannel(
				self.ctx,
				args,
				self.generator,
				self.clientReceivePacketCallback,
				self.ingressSecurityPolicy,
				self.contractStatus,
				self.performanceProfile,
				self.settings,
			)
			if err != nil {
				self.generator.RemoveClientArgs(&args.MultiClientGeneratorClientArgs)
				self.monitor.AddProviderEvent(args.ClientId, ProviderStateEvaluationFailed)
			} else {

				// send an initial ping on the client and let the ack timeout close it
				pingDone, pingCancel := context.WithCancel(self.ctx)
				pendingPingDones = append(pendingPingDones, pingDone)

				// must be called with mutex
				fail := func() {
					select {
					case <-pingDone.Done():
						// already done
						return
					default:
					}

					pingCancel()
					client.Cancel()
					self.generator.RemoveClientArgs(&args.MultiClientGeneratorClientArgs)
					self.monitor.AddProviderEvent(args.ClientId, ProviderStateEvaluationFailed)
				}

				go HandleError(func() {
					mutex.Lock()
					defer mutex.Unlock()

					select {
					case <-pingDone.Done():
						// already done
						return
					default:
					}

					added += 1
					if client.IsP2pOnly() {
						addedP2pOnly += 1
					}

					self.monitor.AddProviderEvent(args.ClientId, ProviderStateInEvaluation)

					success, err := client.SendDetailedMessage(
						&protocol.IpPing{},
						self.settings.PingWriteTimeout,
						func(err error) {
							mutex.Lock()
							defer mutex.Unlock()

							select {
							case <-pingDone.Done():
								// already done
								return
							default:
							}

							if err == nil {
								glog.V(1).Infof("[multi]expand new client\n")

								self.monitor.AddProviderEvent(args.ClientId, ProviderStateAdded)
								var replacedClient *multiClientChannel
								func() {
									self.stateLock.Lock()
									defer self.stateLock.Unlock()
									clientId := client.ClientId()
									replacedClient = self.clients[clientId]
									self.clients[clientId] = client
								}()
								if replacedClient != nil {
									replacedClient.Cancel()
									self.monitor.AddProviderEvent(replacedClient.ClientId(), ProviderStateRemoved)
								}
								pingSuccess += 1
								pingCancel()
							} else {
								glog.V(1).Infof("[multi]create ping error = %s\n", err)
								fail()
							}
						},
					)
					if err != nil {
						glog.Infof("[multi]create client ping error = %s\n", err)
						fail()
					} else if !success {
						fail()
					} else {
						// async wait for the ping
						go HandleError(func() {
							select {
							case <-pingDone.Done():
							case <-time.After(self.settings.PingTimeout):
								glog.V(2).Infof("[multi]expand window timeout waiting for ping\n")
								func() {
									mutex.Lock()
									defer mutex.Unlock()
									fail()
								}()
							}
						}, client.Cancel)
					}
				})
			}
		case <-time.After(timeout):
			glog.V(2).Infof("[multi]expand window timeout waiting for args\n")
		}
	}

	// wait for pending pings
	for _, pingDone := range pendingPingDones {
		timeout := endTime.Sub(time.Now())
		if timeout <= 0 {
			break
		}

		select {
		case <-self.ctx.Done():
			return
		case <-pingDone.Done():
		case <-time.After(timeout):
		}
	}

	return
}

func (self *multiClientWindow) shuffle() {
	for _, client := range self.unorderedClients() {
		client.Cancel()
	}
	self.resizeMonitor.NotifyAll()
}

func (self *multiClientWindow) unorderedClients() []*multiClientChannel {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()
	return maps.Values(self.clients)
}

func (self *multiClientWindow) OrderedClients() []*multiClientChannel {
	var windowSize WindowSizeSettings
	func() {
		self.stateLock.Lock()
		defer self.stateLock.Unlock()
		if self.performanceProfile != nil {
			windowSize = self.performanceProfile.WindowSize
		} else {
			windowSize = self.settings.WindowSizes[self.windowType]
		}
	}()

	clients := []*multiClientChannel{}
	lruTimes := map[*multiClientChannel]time.Time{}
	weights := map[*multiClientChannel]float32{}

	for _, client := range self.unorderedClients() {
		if stats, err := client.WindowStats(); err == nil && !client.isWarning() {
			clients = append(clients, client)
			if !stats.lastEventTime.IsZero() {
				lruTimes[client] = stats.lastEventTime
			}
			weights[client] = float32(1 + stats.ExpectedByteCountPerSecond())
		}
	}

	if 0 == len(clients) {
		return clients
	}

	if glog.V(1) {
		self.statsSampleWeights(weights)
	}

	if 0 < windowSize.FixedWindowSize && windowSize.FixedWindowSize < len(clients) {
		slices.SortFunc(clients, func(a *multiClientChannel, b *multiClientChannel) int {
			lruTimeA := lruTimes[a]
			lruTimeB := lruTimes[b]
			// descending
			if lruTimeA.Before(lruTimeB) {
				return 1
			} else if lruTimeB.Before(lruTimeA) {
				return -1
			}
			weightA := weights[a]
			weightB := weights[b]
			// descending
			if weightA < weightB {
				return 1
			} else if weightB < weightA {
				return -1
			}
			return 0
		})
		clients = clients[:windowSize.FixedWindowSize]
	}

	WeightedShuffleWithEntropy(clients, weights, self.settings.StatsWindowEntropy)

	// use only clients in the min tier
	// this prevents the window from crossing rank until necessary
	minTierClients := []*multiClientChannel{}
	minTier := clients[0].Tier()
	for _, client := range clients[1:] {
		minTier = min(minTier, client.Tier())
	}
	for _, client := range clients {
		if client.Tier() == minTier {
			minTierClients = append(minTierClients, client)
		} else {
			glog.V(1).Infof("[multi]exclude tier from window %d>%d\n", client.Tier(), minTier)
		}
	}

	// use only the top n items from the window
	// if 0 < windowSize.WindowSizeUseMax {
	// 	minTierClients = minTierClients[:min(len(minTierClients), windowSize.WindowSizeUseMax)]
	// }

	return minTierClients
}

func (self *multiClientWindow) statsSampleWeights(weights map[*multiClientChannel]float32) {
	// randonly sample log statistics for weights
	if mathrand.Intn(self.settings.StatsSampleWeightsCount) == 0 {
		// sample the weights
		weightValues := maps.Values(weights)
		slices.SortFunc(weightValues, func(a float32, b float32) int {
			// descending
			if a < b {
				return 1
			} else if b < a {
				return -1
			} else {
				return 0
			}
		})
		net := float32(0)
		for _, weight := range weightValues {
			net += weight
		}
		if 0 < net {
			var sb strings.Builder
			netThresh := float32(0.99)
			netp := float32(0)
			netCount := 0
			for i, weight := range weightValues {
				p := 100 * weight / net
				netp += p
				netCount += 1
				if 0 < i {
					sb.WriteString(" ")
				}
				sb.WriteString(fmt.Sprintf("[%d]%.2f", i, p))
				if netThresh*100 <= netp {
					break
				}
			}

			glog.Infof("[multi]sample weights: %s (+%d more in window <%.0f%%)\n", sb.String(), len(weights)-netCount, 100*(1-netThresh))
		} else {
			glog.Infof("[multi]sample weights: zero (%d in window)\n", len(weights))
		}
	}
}

func (self *multiClientWindow) Close() {
	var removedClients []*multiClientChannel
	func() {
		self.stateLock.Lock()
		defer self.stateLock.Unlock()
		for _, client := range self.clients {
			// client.Close()
			removedClients = append(removedClients, client)
		}
		clear(self.clients)
	}()
	for _, client := range removedClients {
		client.Close()
	}
	// self.removeClients(removedClients)
}

func (self *multiClientWindow) removeClients(removedClients ...*multiClientChannel) {
	for _, client := range removedClients {
		self.monitor.AddProviderEvent(client.ClientId(), ProviderStateRemoved)
	}
	for _, client := range removedClients {
		self.clientRemoveCallback(client)
	}
}

type multiClientChannelArgs struct {
	MultiClientGeneratorClientArgs

	Destination MultiHopId
	DestinationStats
}

type multiClientEventType int

const (
	multiClientEventTypeAck    multiClientEventType = 1
	multiClientEventTypeNack   multiClientEventType = 2
	multiClientEventTypeError  multiClientEventType = 3
	multiClientEventTypeSource multiClientEventType = 4
)

type multiClientEventBucket struct {
	createTime time.Time
	eventTime  time.Time

	sendAckCount        int
	sendAckByteCount    ByteCount
	sendNackCount       int
	sendNackByteCount   ByteCount
	sendSynCount        int
	receiveAckCount     int
	receiveAckByteCount ByteCount
	receiveSynCount     int
	sendAckTime         time.Time
	sendNackTime        time.Time
	sendSynTime         time.Time
	errs                []error
	ip4Paths            map[Ip4Path]bool
	ip6Paths            map[Ip6Path]bool
}

func newMultiClientEventBucket() *multiClientEventBucket {
	now := time.Now()
	return &multiClientEventBucket{
		createTime: now,
		eventTime:  now,
	}
}

type clientWindowStats struct {
	sourceCount                 int
	netSourceCount              int
	sendAckCount                int
	sendAckByteCount            ByteCount
	sendNackCount               int
	sendNackByteCount           ByteCount
	sendSynCount                int
	receiveAckCount             int
	receiveAckByteCount         ByteCount
	receiveSynCount             int
	ackByteCount                ByteCount
	windowDuration              time.Duration
	firstSendAckTime            time.Time
	firstSendNackTime           time.Time
	firstSendSynTime            time.Time
	estimatedByteCountPerSecond ByteCount
	// FIXME firstStatDuration
	clientDuration       time.Duration
	healthyDuration      time.Duration
	unhealthyDuration    time.Duration
	netHealthyDuration   time.Duration
	netUnhealthyDuration time.Duration
	healthy              bool
	removeTime           time.Time
	lastEventTime        time.Time

	// internal
	bucketCount int
}

func (self *clientWindowStats) EffectiveByteCountPerSecond() ByteCount {
	millis := int64(self.windowDuration / time.Millisecond)
	if millis <= 0 {
		return ByteCount(0)
	}
	netByteCount := int64(self.sendAckByteCount + self.receiveAckByteCount)
	return ByteCount((1000*netByteCount + millis/2) / millis)
}

func (self *clientWindowStats) EffectiveByteCount() (send ByteCount, receive ByteCount) {
	millis := int64(self.windowDuration / time.Millisecond)
	if millis <= 0 {
		return
	}
	send = self.sendAckByteCount
	receive = self.receiveAckByteCount
	return
}

func (self *clientWindowStats) ExpectedByteCountPerSecond() ByteCount {
	millis := int64(self.windowDuration / time.Millisecond)
	if millis <= 0 {
		return self.estimatedByteCountPerSecond
	}
	netByteCount := int64(self.sendAckByteCount + self.sendNackByteCount + self.receiveAckByteCount)
	glog.V(2).Infof("[multi]expected use estimated = %dbps (net = %db/%dms)\n", self.estimatedByteCountPerSecond, netByteCount, millis)
	return max(
		self.estimatedByteCountPerSecond-ByteCount((1000*netByteCount+millis/2)/millis),
		0,
	)
}

type multiClientChannel struct {
	ctx    context.Context
	cancel context.CancelFunc

	args *multiClientChannelArgs

	api *BringYourApi

	clientReceivePacketCallback clientReceivePacketFunction
	ingressSecurityPolicy       SecurityPolicy
	performanceProfile          *PerformanceProfile
	createTime                  time.Time

	settings *MultiClientSettings

	// sourceFilter map[TransferPath]bool

	client *Client

	stateLock    sync.Mutex
	eventBuckets []*multiClientEventBucket
	// destination -> source -> count
	ip4DestinationSourceCount          map[Ip4Path]map[Ip4Path]int
	ip6DestinationSourceCount          map[Ip6Path]map[Ip6Path]int
	packetStats                        *clientWindowStats
	endErr                             error
	maxEffectiveByteCountPerSecond     ByteCount
	maxEffectiveByteCountPerSecondTime time.Time
	firstEventTime                     time.Time

	healthy              bool
	lastHealthyTime      time.Time
	lastUnhealthyTime    time.Time
	netHealthyDuration   time.Duration
	netUnhealthyDuration time.Duration

	// affinityCount int
	// affinityTime  time.Time

	clientReceiveUnsub func()

	warning bool
}

func newMultiClientChannel(
	ctx context.Context,
	args *multiClientChannelArgs,
	generator MultiClientGenerator,
	clientReceivePacketCallback clientReceivePacketFunction,
	ingressSecurityPolicy SecurityPolicy,
	contractStatusCallback ContractStatusFunction,
	performanceProfile *PerformanceProfile,
	settings *MultiClientSettings,
) (*multiClientChannel, error) {
	cancelCtx, cancel := context.WithCancel(ctx)

	clientSettings := generator.NewClientSettings()
	clientSettings.SendBufferSettings.AckTimeout = settings.AckTimeout

	client, err := generator.NewClient(
		cancelCtx,
		&args.MultiClientGeneratorClientArgs,
		clientSettings,
	)
	if err != nil {
		return nil, err
	}
	contractStatusSub := client.ContractManager().AddContractStatusCallback(contractStatusCallback)
	go HandleError(func() {
		select {
		case <-cancelCtx.Done():
		case <-client.Done():
		}
		client.Cancel()
		contractStatusSub()
		generator.RemoveClientWithArgs(client, &args.MultiClientGeneratorClientArgs)
	}, cancel)

	// sourceFilter := map[TransferPath]bool{
	//     Path{ClientId:args.DestinationId}: true,
	// }

	clientChannel := &multiClientChannel{
		ctx:                         cancelCtx,
		cancel:                      cancel,
		args:                        args,
		clientReceivePacketCallback: clientReceivePacketCallback,
		ingressSecurityPolicy:       ingressSecurityPolicy,
		performanceProfile:          performanceProfile,
		createTime:                  time.Now(),
		settings:                    settings,
		// sourceFilter: sourceFilter,
		client:                    client,
		eventBuckets:              []*multiClientEventBucket{},
		ip4DestinationSourceCount: map[Ip4Path]map[Ip4Path]int{},
		ip6DestinationSourceCount: map[Ip6Path]map[Ip6Path]int{},
		packetStats:               &clientWindowStats{},
		// affinityCount:             0,
		// affinityTime:              time.Time{},
	}
	go HandleError(clientChannel.detectBlackhole, cancel)
	go HandleError(clientChannel.ping, cancel)

	clientReceiveUnsub := client.AddReceiveCallback(clientChannel.clientReceive)
	clientChannel.clientReceiveUnsub = clientReceiveUnsub

	return clientChannel, nil
}

func (self *multiClientChannel) ClientId() Id {
	return self.client.ClientId()
}

func (self *multiClientChannel) IsP2pOnly() bool {
	return self.args.MultiClientGeneratorClientArgs.P2pOnly
}

func (self *multiClientChannel) Tier() int {
	return self.args.DestinationStats.Tier
}

func (self *multiClientChannel) EstimatedByteCountPerSecond() ByteCount {
	return self.args.EstimatedBytesPerSecond
}

func (self *multiClientChannel) setWarning(warning bool) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()
	self.warning = warning
}

func (self *multiClientChannel) isWarning() bool {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()
	return self.warning
}

// func (self *multiClientChannel) UpdateAffinity() {
// 	self.stateLock.Lock()
// 	defer self.stateLock.Unlock()

// 	self.affinityCount += 1
// 	self.affinityTime = time.Now()
// }

// func (self *multiClientChannel) ClearAffinity() {
// 	self.stateLock.Lock()
// 	defer self.stateLock.Unlock()

// 	self.affinityCount = 0
// 	self.affinityTime = time.Time{}
// }

// func (self *multiClientChannel) MostRecentAffinity() (int, time.Time) {
// 	self.stateLock.Lock()
// 	defer self.stateLock.Unlock()

// 	return self.affinityCount, self.affinityTime
// }

func (self *multiClientChannel) Send(parsedPacket *parsedPacket, timeout time.Duration) bool {
	success, err := self.SendDetailed(parsedPacket, timeout)
	return success && err == nil
}

func (self *multiClientChannel) SendDetailed(parsedPacket *parsedPacket, timeout time.Duration) (bool, error) {
	var ack bool
	switch parsedPacket.ipPath.Protocol {
	case IpProtocolUdp:
		if self.settings.UdpCollapsePrevention {
			ack = false
		} else {
			ack = true
		}
	default:
		ack = true
	}
	return self.SendDetailedWithAck(parsedPacket, timeout, ack)
}

func (self *multiClientChannel) SendWithAck(parsedPacket *parsedPacket, timeout time.Duration, ack bool) bool {
	success, err := self.SendDetailedWithAck(parsedPacket, timeout, ack)
	return success && err == nil
}

func (self *multiClientChannel) SendDetailedWithAck(parsedPacket *parsedPacket, timeout time.Duration, ack bool) (bool, error) {
	ipPacketToProvider := &protocol.IpPacketToProvider{
		IpPacket: &protocol.IpPacket{
			PacketBytes: parsedPacket.packet,
		},
	}
	if frame, err := ToFrame(ipPacketToProvider, self.settings.ProtocolVersion); err != nil {
		self.addError(err)
		return false, err
	} else {
		packetByteCount := ByteCount(len(parsedPacket.packet))
		self.addSendNack(packetByteCount)
		if parsedPacket.ipPath.Syn {
			self.addSendSyn(1)
		}
		self.addSource(parsedPacket.ipPath)
		ackCallback := func(err error) {
			if err == nil {
				self.addSendAck(packetByteCount)
			} else {
				self.addError(err)
			}
		}

		var opts []any
		if self.performanceProfile != nil && self.performanceProfile.AllowDirect {
			opts = append(opts, ForceStream())
		}
		if !ack {
			opts = append(opts, NoAck())
		}
		success, err := self.client.SendMultiHopWithTimeoutDetailed(
			frame,
			self.args.Destination,
			ackCallback,
			timeout,
			opts...,
		)
		if err != nil {
			return success, err
		}
		if success {
			if !frame.Raw {
				MessagePoolReturn(parsedPacket.packet)
			}
		} else {
			if !frame.Raw {
				MessagePoolReturn(frame.MessageBytes)
			}
		}
		return success, err
	}
}

func (self *multiClientChannel) SendDetailedMessage(message proto.Message, timeout time.Duration, ackCallback func(error)) (bool, error) {
	if frame, err := ToFrame(message, self.settings.ProtocolVersion); err != nil {
		return false, err
	} else {
		var opts []any
		if self.performanceProfile != nil && self.performanceProfile.AllowDirect {
			opts = append(opts, ForceStream())
		}
		return self.client.SendMultiHopWithTimeoutDetailed(
			frame,
			self.args.Destination,
			ackCallback,
			timeout,
			opts...,
		)
	}
}

func (self *multiClientChannel) Done() <-chan struct{} {
	return self.ctx.Done()
}

func (self *multiClientChannel) Destination() MultiHopId {
	return self.args.Destination
}

func (self *multiClientChannel) detectBlackhole() {
	// within a timeout window, if there are sent data but none received,
	// error out. This is similar to an ack timeout.
	defer self.cancel()

	for {
		if windowStats, err := self.WindowStats(); err != nil {
			return
		} else {
			blackhole := func() bool {
				now := time.Now()
				if !windowStats.firstSendNackTime.IsZero() && self.settings.BlackholeTimeout-now.Sub(windowStats.firstSendNackTime) <= 0 {
					if windowStats.sendAckCount <= 0 {
						return true
					}
					if windowStats.receiveAckCount <= 0 {
						return true
					}
				}
				if !windowStats.firstSendSynTime.IsZero() && self.settings.BlackholeConnectTimeout-now.Sub(windowStats.firstSendSynTime) <= 0 {
					if windowStats.receiveSynCount <= 0 {
						return true
					}
				}
				return false
			}()

			if blackhole {
				// the client has sent data but received nothing back
				// this looks like a blackhole
				glog.V(1).Infof("[multi]routing %s blackhole: %d %dB <> %d %dB (%d <> %d)\n",
					self.args.Destination,
					windowStats.sendAckCount,
					windowStats.sendAckByteCount,
					windowStats.receiveAckCount,
					windowStats.receiveAckByteCount,
					windowStats.sendSynCount,
					windowStats.receiveSynCount,
				)
				self.addError(fmt.Errorf("Blackhole (%d %dB)",
					windowStats.sendAckCount,
					windowStats.sendAckByteCount,
				))
				return
			} else {
				glog.V(1).Infof(
					"[multi]routing ok %s: %d %dB <> %d %dB (%d <> %d)\n",
					self.args.Destination,
					windowStats.sendAckCount,
					windowStats.sendAckByteCount,
					windowStats.receiveAckCount,
					windowStats.receiveAckByteCount,
					windowStats.sendSynCount,
					windowStats.receiveSynCount,
				)
			}

			select {
			case <-self.ctx.Done():
				return
			case <-self.client.Done():
				return
			case <-time.After(self.settings.BlackholeTimeout / 4):
			}
		}
	}
}

func (self *multiClientChannel) ping() {
	defer self.cancel()

	for {
		if windowStats, err := self.WindowStats(); err != nil {
			return
		} else if self.settings.CPingMaxByteCountPerSecond == 0 || windowStats.EffectiveByteCountPerSecond() <= self.settings.CPingMaxByteCountPerSecond {
			pingDone := make(chan error)
			success, err := self.SendDetailedMessage(
				&protocol.IpPing{},
				self.settings.CPingWriteTimeout,
				func(err error) {
					defer close(pingDone)
					select {
					case <-self.ctx.Done():
						return
					case pingDone <- err:
					}
				},
			)
			if err != nil {
				close(pingDone)
				return
			} else if !success {
				close(pingDone)
				return
			} else {
				select {
				case <-self.ctx.Done():
					return
				case <-self.client.Done():
					return
				case err := <-pingDone:
					if err != nil {
						self.addError(err)
						return
					}
				case <-time.After(self.settings.CPingTimeout):
					return
				}
			}
		}

		select {
		case <-self.ctx.Done():
			return
		case <-self.client.Done():
			return
		case <-WakeupAfter(self.settings.CPingTimeout, self.settings.CPingTimeout):
		}
	}
}

func (self *multiClientChannel) addSendNack(ackByteCount ByteCount) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	self.packetStats.sendNackCount += 1
	self.packetStats.sendNackByteCount += ackByteCount

	eventBucket := self.eventBucket()
	if eventBucket.sendNackCount == 0 {
		eventBucket.sendNackTime = time.Now()
	}
	eventBucket.sendNackCount += 1
	eventBucket.sendNackByteCount += ackByteCount
}

func (self *multiClientChannel) addSendAck(ackByteCount ByteCount) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	self.packetStats.sendNackCount -= 1
	self.packetStats.sendNackByteCount -= ackByteCount
	self.packetStats.sendAckCount += 1
	self.packetStats.sendAckByteCount += ackByteCount

	eventBucket := self.eventBucket()
	if eventBucket.sendAckCount == 0 {
		eventBucket.sendAckTime = time.Now()
	}
	eventBucket.sendAckCount += 1
	eventBucket.sendAckByteCount += ackByteCount
}

func (self *multiClientChannel) addSendSyn(synCount int) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	self.packetStats.sendSynCount += synCount

	eventBucket := self.eventBucket()
	if eventBucket.sendSynCount == 0 {
		eventBucket.sendSynTime = time.Now()
	}
	eventBucket.sendSynCount += synCount
}

func (self *multiClientChannel) addReceiveAck(ackByteCount ByteCount) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	self.packetStats.receiveAckCount += 1
	self.packetStats.receiveAckByteCount += ackByteCount

	eventBucket := self.eventBucket()
	eventBucket.receiveAckCount += 1
	eventBucket.receiveAckByteCount += ackByteCount
}

func (self *multiClientChannel) addReceiveSyn(synCount int) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	self.packetStats.receiveSynCount += synCount

	eventBucket := self.eventBucket()
	eventBucket.receiveSynCount += synCount
}

func (self *multiClientChannel) addSource(ipPath *IpPath) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	eventBucket := self.eventBucket()
	switch ipPath.Version {
	case 4:
		ip4Path := ipPath.ToIp4Path()

		if eventBucket.ip4Paths == nil {
			eventBucket.ip4Paths = map[Ip4Path]bool{}
		}
		eventBucket.ip4Paths[ip4Path] = true

		source := ip4Path.Source()
		destination := ip4Path.Destination()

		sourceCount, ok := self.ip4DestinationSourceCount[destination]
		if !ok {
			sourceCount = map[Ip4Path]int{}
			self.ip4DestinationSourceCount[destination] = sourceCount
		}
		sourceCount[source] += 1
	case 6:
		ip6Path := ipPath.ToIp6Path()

		if eventBucket.ip6Paths == nil {
			eventBucket.ip6Paths = map[Ip6Path]bool{}
		}
		eventBucket.ip6Paths[ip6Path] = true

		source := ip6Path.Source()
		destination := ip6Path.Destination()

		sourceCount, ok := self.ip6DestinationSourceCount[destination]
		if !ok {
			sourceCount = map[Ip6Path]int{}
			self.ip6DestinationSourceCount[destination] = sourceCount
		}
		sourceCount[source] += 1
	default:
		panic(fmt.Errorf("Bad protocol version %d", ipPath.Version))
	}
}

func (self *multiClientChannel) addError(err error) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	if self.endErr == nil {
		self.endErr = err
	}

	eventBucket := self.eventBucket()
	eventBucket.errs = append(eventBucket.errs, err)
}

// must be called with `stateLock`
func (self *multiClientChannel) eventBucket() *multiClientEventBucket {
	now := time.Now()

	var eventBucket *multiClientEventBucket
	if n := len(self.eventBuckets); 0 < n {
		eventBucket = self.eventBuckets[n-1]
	}

	if eventBucket == nil || eventBucket.createTime.Add(self.settings.StatsWindowBucketDuration).Before(now) {
		eventBucket = newMultiClientEventBucket()
		self.eventBuckets = append(self.eventBuckets, eventBucket)
	}

	eventBucket.eventTime = now

	self.coalesceEventBuckets()

	return eventBucket
}

// must be called with `stateLock`
func (self *multiClientChannel) coalesceEventBuckets() {
	// if there is no activity (no new buckets), keep historical buckets around
	minBucketCount := 1 + int(self.settings.StatsWindowDuration/self.settings.StatsWindowBucketDuration)

	windowStart := time.Now().Add(-self.settings.StatsWindowDuration)

	removeEventBucket := func(eventBucket *multiClientEventBucket) {
		self.packetStats.sendAckCount -= eventBucket.sendAckCount
		self.packetStats.sendAckByteCount -= eventBucket.sendAckByteCount
		self.packetStats.sendSynCount -= eventBucket.sendSynCount
		self.packetStats.receiveAckCount -= eventBucket.receiveAckCount
		self.packetStats.receiveAckByteCount -= eventBucket.receiveAckByteCount
		self.packetStats.receiveSynCount -= eventBucket.receiveSynCount

		for ip4Path, _ := range eventBucket.ip4Paths {
			source := ip4Path.Source()
			destination := ip4Path.Destination()

			sourceCount, ok := self.ip4DestinationSourceCount[destination]
			if ok {
				count := sourceCount[source]
				if count-1 <= 0 {
					delete(sourceCount, source)
				} else {
					sourceCount[source] = count - 1
				}
				if len(sourceCount) == 0 {
					delete(self.ip4DestinationSourceCount, destination)
				}
			}
		}

		for ip6Path, _ := range eventBucket.ip6Paths {
			source := ip6Path.Source()
			destination := ip6Path.Destination()

			sourceCount, ok := self.ip6DestinationSourceCount[destination]
			if ok {
				count := sourceCount[source]
				if count-1 <= 0 {
					delete(sourceCount, source)
				} else {
					sourceCount[source] = count - 1
				}
				if len(sourceCount) == 0 {
					delete(self.ip6DestinationSourceCount, destination)
				}
			}
		}
	}

	// remove all events before the window start
	i := 0
	for i < len(self.eventBuckets) && self.eventBuckets[i].eventTime.Before(windowStart) {
		removeEventBucket(self.eventBuckets[i])
		self.eventBuckets[i] = nil
		i += 1
	}
	for i < len(self.eventBuckets) && minBucketCount < len(self.eventBuckets) {
		removeEventBucket(self.eventBuckets[i])
		self.eventBuckets[i] = nil
		i += 1
	}
	if 0 < i {
		self.eventBuckets = self.eventBuckets[i:]
	}
}

func (self *multiClientChannel) WindowStats() (*clientWindowStats, error) {
	return self.windowStatsWithCoalesce(true)
}

func (self *multiClientChannel) windowStatsWithCoalesce(coalesce bool) (*clientWindowStats, error) {
	self.stateLock.Lock()
	defer self.stateLock.Unlock()

	if coalesce {
		self.coalesceEventBuckets()
	}

	// omit the latest two event buckets since they may be partial
	var eventBuckets []*multiClientEventBucket
	if 2 <= len(self.eventBuckets) {
		eventBuckets = self.eventBuckets[0 : len(self.eventBuckets)-2]
	}

	windowDuration := time.Duration(0)
	if 0 < len(eventBuckets) {
		endTime := eventBuckets[len(eventBuckets)-1].eventTime
		windowDuration = endTime.Sub(eventBuckets[0].createTime)
	}
	var firstSendAckTime time.Time
	for _, eventBucket := range eventBuckets {
		if 0 < eventBucket.sendAckCount {
			firstSendAckTime = eventBucket.sendAckTime
			break
		}
	}
	var firstSendNackTime time.Time
	for _, eventBucket := range eventBuckets {
		if 0 < eventBucket.sendNackCount {
			firstSendNackTime = eventBucket.sendNackTime
			break
		}
	}
	var firstSendSynTime time.Time
	for _, eventBucket := range eventBuckets {
		if 0 < eventBucket.sendSynCount {
			firstSendSynTime = eventBucket.sendSynTime
			break
		}
	}

	// public internet resource ports
	isPublicPort := func(port int) bool {
		switch port {
		case 443:
			return true
		default:
			return false
		}
	}

	netSourceCounts := []int{}
	for ip4Path, sourceCounts := range self.ip4DestinationSourceCount {
		if isPublicPort(ip4Path.DestinationPort) {
			netSourceCounts = append(netSourceCounts, len(sourceCounts))
		}
	}
	for ip6Path, sourceCounts := range self.ip6DestinationSourceCount {
		if isPublicPort(ip6Path.DestinationPort) {
			netSourceCounts = append(netSourceCounts, len(sourceCounts))
		}
	}
	slices.Sort(netSourceCounts)
	maxSourceCount := 0
	selectionIndex := int(math.Ceil(
		self.settings.StatsSourceCountSelection * float64(len(netSourceCounts)-1),
	))
	if selectionIndex < len(netSourceCounts) {
		maxSourceCount = netSourceCounts[selectionIndex]
	}
	netSourceCount := 0
	for _, sourceCounts := range self.ip4DestinationSourceCount {
		netSourceCount += len(sourceCounts)
	}
	for _, sourceCounts := range self.ip6DestinationSourceCount {
		netSourceCount += len(sourceCounts)
	}
	if glog.V(2) {
		for ip4Path, sourceCounts := range self.ip4DestinationSourceCount {
			if isPublicPort(ip4Path.DestinationPort) {
				if len(sourceCounts) == maxSourceCount {
					glog.Infof("[multi]max source count %d = %v\n", maxSourceCount, ip4Path)
				}
			}
		}
		for ip6Path, sourceCounts := range self.ip6DestinationSourceCount {
			if isPublicPort(ip6Path.DestinationPort) {
				if len(sourceCounts) == maxSourceCount {
					glog.Infof("[multi]max source count %d = %v\n", maxSourceCount, ip6Path)
				}
			}
		}
	}

	stats := &clientWindowStats{
		sourceCount:         maxSourceCount,
		netSourceCount:      netSourceCount,
		sendAckCount:        self.packetStats.sendAckCount,
		sendNackCount:       self.packetStats.sendNackCount,
		sendAckByteCount:    self.packetStats.sendAckByteCount,
		sendSynCount:        self.packetStats.sendSynCount,
		sendNackByteCount:   self.packetStats.sendNackByteCount,
		receiveAckCount:     self.packetStats.receiveAckCount,
		receiveAckByteCount: self.packetStats.receiveAckByteCount,
		receiveSynCount:     self.packetStats.receiveSynCount,
		windowDuration:      windowDuration,
		firstSendAckTime:    firstSendAckTime,
		firstSendNackTime:   firstSendNackTime,
		firstSendSynTime:    firstSendSynTime,
		bucketCount:         len(eventBuckets),
	}
	if 0 < len(eventBuckets) || !self.firstEventTime.IsZero() {
		// var eventTime time.Time
		// if 0 < len(eventBuckets) {
		// 	eventTime = eventBuckets[len(eventBuckets)-1].eventTime
		// } else {
		// 	eventTime = time.Now()
		// }
		eventTime := time.Now()

		if 0 < len(eventBuckets) {
			stats.lastEventTime = eventBuckets[len(eventBuckets)-1].eventTime
		}

		effectiveByteCountPerSecond := stats.EffectiveByteCountPerSecond()
		// scaledEffectiveByteCountPerSecond := ByteCount(self.settings.StatsWindowMaxEffectiveByteCountPerSecondScale * float32(stats.EffectiveByteCountPerSecond()))
		if self.maxEffectiveByteCountPerSecond < effectiveByteCountPerSecond {
			self.maxEffectiveByteCountPerSecond = effectiveByteCountPerSecond
			self.maxEffectiveByteCountPerSecondTime = eventTime
		}

		effectiveSendByteCount, effectiveReceiveByteCount := stats.EffectiveByteCount()
		healthy := (0 < effectiveSendByteCount) == (0 < effectiveReceiveByteCount)
		if healthy {
			if self.lastUnhealthyTime.IsZero() {
				self.lastUnhealthyTime = eventTime
			}

			if !self.healthy {
				self.healthy = true
				if !self.lastHealthyTime.IsZero() {
					self.netUnhealthyDuration += self.lastUnhealthyTime.Sub(self.lastHealthyTime)
				}
			}

			self.lastHealthyTime = eventTime

			stats.healthyDuration = eventTime.Sub(self.lastUnhealthyTime)
		} else {
			if self.lastHealthyTime.IsZero() {
				self.lastHealthyTime = eventTime
			}

			if self.healthy {
				self.healthy = false
				if !self.lastUnhealthyTime.IsZero() {
					self.netHealthyDuration += self.lastHealthyTime.Sub(self.lastUnhealthyTime)
				}
			}

			self.lastUnhealthyTime = eventTime

			stats.unhealthyDuration = eventTime.Sub(self.lastHealthyTime)
		}
		stats.healthy = healthy
		stats.netHealthyDuration = self.netHealthyDuration + stats.healthyDuration
		stats.netUnhealthyDuration = self.netUnhealthyDuration + stats.unhealthyDuration
		if self.firstEventTime.IsZero() {
			self.firstEventTime = eventBuckets[0].createTime
		}
		stats.clientDuration = eventTime.Sub(self.firstEventTime)
		stats.removeTime = self.firstEventTime.Add(self.settings.MaxClientLifetime)
	}
	// if !self.firstEventTime.IsZero() {
	// 	stats.removeTime = self.firstEventTime.Add(self.settings.MaxClientLifetime)
	// }
	if self.settings.StatsWindowGraceperiod < stats.clientDuration {
		stats.estimatedByteCountPerSecond = self.maxEffectiveByteCountPerSecond
	} else {
		stats.estimatedByteCountPerSecond = max(
			min(self.EstimatedByteCountPerSecond(), self.settings.StatsWindowMaxEstimatedByteCountPerSecond),
			self.maxEffectiveByteCountPerSecond,
		)
	}

	err := self.endErr
	if err == nil {
		select {
		case <-self.ctx.Done():
			err = errors.New("Done.")
		case <-self.client.Done():
			err = errors.New("Done.")
		default:
		}
	}

	return stats, err
}

// `connect.ReceiveFunction`
func (self *multiClientChannel) clientReceive(source TransferPath, frames []*protocol.Frame, provideMode protocol.ProvideMode) {
	select {
	case <-self.ctx.Done():
		return
	default:
	}

	// only process frames from the destinations
	// if allow := self.sourceFilter[source]; !allow {
	//     glog.V(2).Infof("[multi]receive drop %d %s<-\n", len(frames), self.args.DestinationId)
	//     return
	// }

	for _, frame := range frames {
		switch frame.MessageType {
		case protocol.MessageType_IpIpPacketFromProvider:
			if ipPacketFromProvider_, err := FromFrame(frame); err == nil {
				ipPacketFromProvider := ipPacketFromProvider_.(*protocol.IpPacketFromProvider)

				packet := ipPacketFromProvider.IpPacket.PacketBytes

				ipPath, err := ParseIpPath(packet)
				if err == nil {
					self.addReceiveAck(ByteCount(len(packet)))
					if ipPath.Syn {
						self.addReceiveSyn(1)
					}
					self.clientReceivePacketCallback(self, source, provideMode, ipPath, packet)
				}
				// else not an ip packet, drop
			} else {
				glog.V(2).Infof("[multi]receive drop %s<- = %s\n", self.args.Destination, err)
			}
		default:
			// unknown message, drop
		}
	}
}

func (self *multiClientChannel) Cancel() {
	self.addError(errors.New("Done."))
	self.cancel()
	self.client.Cancel()
}

func (self *multiClientChannel) Close() {
	self.addError(errors.New("Done."))
	self.cancel()
	self.client.Close()

	self.clientReceiveUnsub()
}

func (self *multiClientChannel) IsDone() bool {
	select {
	case <-self.ctx.Done():
		return true
	default:
		return false
	}
}