server+inbound: correct inbound admission accounting

In this commit, we keep loopback and whitelisted peers inside the ordinary
pending-handshake and V2 source budgets, while preserving their existing
no-ban behavior.

We also derive the listener reservation from the configured peer mode.
Connect-only and simnet nodes now reserve just their permanent peers, while
automatic mode accounts for both its target and addpeers without exceeding
MaxPeers.

Finally, a bound V2 handshake consumes its rate budgets once, but reacquires
the concurrency slot for each CPU-bound responder phase. This keeps rate
limiting scoped to the handshake while bounding both expensive phases.

config.go

@@ -129,7 +129,7 @@ type config struct {
 	Listeners            []string      `long:"listen" description:"Add an interface/port to listen for connections (default all interfaces port: 8333, testnet: 18333)"`
 	LogDir               string        `long:"logdir" description:"Directory to log output."`
 	MaxOrphanTxs         int           `long:"maxorphantx" description:"Max number of orphan transactions to keep in memory"`
-	MaxPeers             int           `long:"maxpeers" description:"Max number of inbound and outbound peers. Up to 8 slots are reserved for automatic outbound peers; values of 8 or less disable inbound connections"`
+	MaxPeers             int           `long:"maxpeers" description:"Max number of inbound and outbound peers. Outbound slots for the configured peer mode are reserved before inbound capacity is calculated"`
 	MiningAddrs          []string      `long:"miningaddr" description:"Add the specified payment address to the list of addresses to use for generated blocks -- At least one address is required if the generate option is set"`
 	MinRelayTxFee        float64       `long:"minrelaytxfee" description:"The minimum transaction fee in BTC/kB to be considered a non-zero fee."`
 	DisableBanning       bool          `long:"nobanning" description:"Disable banning of misbehaving peers"`

internal/inbound/admission.go

@@ -100,17 +100,37 @@ type Admission struct {
 }
 
 // V2Admission binds the server-wide v2 admission policy to a single remote
-// address. The transport uses this value after it has classified the
-// connection as v2, but before it performs key generation or key agreement.
+// address. Its first successful acquisition consumes the handshake's rate
+// budgets. Each acquisition reserves a fresh concurrency slot for one
+// CPU-bound responder phase.
 type V2Admission struct {
 	admission          *Admission
 	remote             net.Addr
 	bypassSourceLimits bool
+
+	mu           sync.Mutex
+	rateAdmitted bool
 }
 
-// Acquire reserves the v2 rate and concurrency budgets for the bound remote.
+// Acquire reserves one v2 concurrency slot for the bound remote. The first
+// successful call also consumes the global and per-source rate budgets.
 func (a *V2Admission) Acquire() (func(), error) {
-	return a.admission.admitV2(a.remote, a.bypassSourceLimits)
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	if a.rateAdmitted {
+		return a.admission.acquireV2Slot(a.remote)
+	}
+
+	release, err := a.admission.admitV2(
+		a.remote, a.bypassSourceLimits,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	a.rateAdmitted = true
+	return release, nil
 }
 
 // BindV2 binds the v2 admission policy to a remote address.
@@ -279,8 +299,8 @@ func (a *Admission) AcquireSource(
 	return release, nil
 }
 
-// admitV2 reserves the rate and concurrency budgets for the CPU-bound portion
-// of an inbound v2 handshake.  The returned release function only releases the
+// admitV2 reserves the one-time rate budgets and the first concurrency slot
+// for an inbound v2 handshake. The returned release function only releases the
 // concurrency slot; consumed rate tokens are not returned.
 func (a *Admission) admitV2(
 	addr net.Addr, bypassSourceLimits bool,
@@ -311,6 +331,22 @@ func (a *Admission) admitV2(
 		return nil, errV2HandshakeRateLimit
 	}
 
+	release, err := a.acquireV2Slot(addr)
+	if err != nil {
+		globalReservation.CancelAt(now)
+		if sourceReservation != nil {
+			sourceReservation.CancelAt(now)
+		}
+
+		return nil, err
+	}
+
+	return release, nil
+}
+
+// acquireV2Slot reserves one concurrency slot for a CPU-bound responder
+// phase. It does not consume a handshake rate token.
+func (a *Admission) acquireV2Slot(addr net.Addr) (func(), error) {
 	select {
 	case a.v2Slots <- struct{}{}:
 		var once sync.Once
@@ -321,10 +357,6 @@ func (a *Admission) admitV2(
 		}, nil
 
 	default:
-		globalReservation.CancelAt(now)
-		if sourceReservation != nil {
-			sourceReservation.CancelAt(now)
-		}
 		a.logV2Rejection(addr, "concurrency")
 		return nil, errV2HandshakeConcurrency
 	}

internal/inbound/admission_test.go

@@ -299,6 +299,87 @@ func TestV2HandshakeConcurrency(t *testing.T) {
 	replacement()
 }
 
+// TestV2HandshakePhases verifies a bound handshake consumes its rate budgets
+// once while each CPU phase reacquires the concurrency slot.
+func TestV2HandshakePhases(t *testing.T) {
+	t.Parallel()
+
+	now := time.Unix(1000, 0)
+	remote := &net.TCPAddr{IP: net.ParseIP("192.0.2.1"), Port: 8333}
+	other := &net.TCPAddr{IP: net.ParseIP("192.0.3.1"), Port: 8333}
+	admission := newAdmission(admissionConfig{
+		maxPendingPerSource: 1,
+		v2Rate:              0,
+		v2Burst:             1,
+		v2SourceRate:        0,
+		v2SourceBurst:       1,
+		v2SourceCacheSize:   16,
+		v2Concurrency:       1,
+		now:                 func() time.Time { return now },
+	})
+
+	bound := admission.BindV2(remote, false)
+	releaseFirst, err := bound.Acquire()
+	require.NoError(t, err)
+
+	_, err = bound.Acquire()
+	require.ErrorIs(t, err, errV2HandshakeConcurrency,
+		"each phase must reserve a fresh concurrency slot")
+
+	releaseFirst()
+	releaseFirst()
+
+	releaseSecond, err := bound.Acquire()
+	require.NoError(t, err,
+		"the second phase must not consume another rate token")
+	releaseSecond()
+	releaseSecond()
+
+	_, err = admission.BindV2(other, false).Acquire()
+	require.ErrorIs(t, err, errV2HandshakeRateLimit,
+		"a new handshake must consume a new global rate token")
+
+	_, err = admission.BindV2(remote, false).Acquire()
+	require.ErrorIs(t, err, errV2HandshakeSourceRateLimit,
+		"a new handshake must consume a new source rate token")
+}
+
+// TestV2HandshakeFirstAcquireRetry verifies a concurrency rejection leaves a
+// bound admission in its first-acquire state.
+func TestV2HandshakeFirstAcquireRetry(t *testing.T) {
+	t.Parallel()
+
+	now := time.Unix(1000, 0)
+	sourceA := &net.TCPAddr{IP: net.ParseIP("192.0.2.1"), Port: 1}
+	sourceB := &net.TCPAddr{IP: net.ParseIP("192.0.3.1"), Port: 2}
+	admission := newAdmission(admissionConfig{
+		maxPendingPerSource: 1,
+		v2Rate:              rate.Inf,
+		v2SourceRate:        0,
+		v2SourceBurst:       1,
+		v2SourceCacheSize:   16,
+		v2Concurrency:       1,
+		now:                 func() time.Time { return now },
+	})
+
+	releaseA, err := admission.BindV2(sourceA, false).Acquire()
+	require.NoError(t, err)
+
+	boundB := admission.BindV2(sourceB, false)
+	_, err = boundB.Acquire()
+	require.ErrorIs(t, err, errV2HandshakeConcurrency)
+	releaseA()
+
+	releaseB, err := admission.BindV2(sourceB, false).Acquire()
+	require.NoError(t, err,
+		"a concurrency rejection must return the source rate token")
+	releaseB()
+
+	_, err = boundB.Acquire()
+	require.ErrorIs(t, err, errV2HandshakeSourceRateLimit,
+		"a rejected first acquisition must retry rate admission")
+}
+
 // TestV2HandshakeGlobalRateRollback verifies a global rejection does not
 // consume the rejected source's independent token.
 func TestV2HandshakeGlobalRateRollback(t *testing.T) {

peer/peer_test.go

@@ -1301,8 +1301,8 @@ func TestSendAddrV2Handshake(t *testing.T) {
 	}
 }
 
-// TestV2HandshakeAdmission verifies the responder admission hook is invoked
-// exactly once for an inbound v2 handshake and never for the initiator.
+// TestV2HandshakeAdmission verifies the responder admission hook bounds both
+// inbound CPU phases and is never invoked for the initiator.
 func TestV2HandshakeAdmission(t *testing.T) {
 	verack := make(chan struct{}, 2)
 	var (
@@ -1355,10 +1355,10 @@ func TestV2HandshakeAdmission(t *testing.T) {
 		}
 	}
 
-	if got := admissions.Load(); got != 1 {
-		t.Fatalf("admission invoked %d times, want 1", got)
+	if got := admissions.Load(); got != 2 {
+		t.Fatalf("admission invoked %d times, want 2", got)
 	}
-	if got := releases.Load(); got != 1 {
-		t.Fatalf("admission released %d times, want 1", got)
+	if got := releases.Load(); got != 2 {
+		t.Fatalf("admission released %d times, want 2", got)
 	}
 }

server.go

@@ -74,14 +74,32 @@ var (
 // zeroHash is the zero value hash (all zeros).  It is defined as a convenience.
 var zeroHash chainhash.Hash
 
+// reservedOutboundPeers returns the outbound connection reservation for the
+// configured peer mode, capped at the total peer limit.
+func reservedOutboundPeers(
+	maxPeers, targetOutbound, permanentPeers int, automaticOutbound bool,
+) int {
+
+	reserved := permanentPeers
+	if automaticOutbound {
+		reserved += targetOutbound
+	}
+
+	if reserved > maxPeers {
+		return maxPeers
+	}
+
+	return reserved
+}
+
 // maxInboundPeers returns the accepted inbound connection budget after
-// reserving capacity for automatic outbound peers.
-func maxInboundPeers(maxPeers, targetOutbound int) uint32 {
-	if maxPeers <= targetOutbound {
+// reserving capacity for outbound peers.
+func maxInboundPeers(maxPeers, reservedOutbound int) uint32 {
+	if maxPeers <= reservedOutbound {
 		return 0
 	}
 
-	return uint32(maxPeers - targetOutbound)
+	return uint32(maxPeers - reservedOutbound)
 }
 
 // onionAddr implements the net.Addr interface and represents a tor address.
@@ -2306,29 +2324,39 @@ func newPeerConfig(sp *serverPeer) *peer.Config {
 	}
 }
 
+// acquireInboundPeerAdmission reserves the source budgets for an inbound peer
+// and reports whether the peer retains the existing no-ban permission.
+func (s *server) acquireInboundPeerAdmission(
+	remoteAddr net.Addr,
+) (bool, func(), *inbound.V2Admission, error) {
+
+	whitelisted := isWhitelisted(remoteAddr)
+	if s.inboundAdmission == nil {
+		return whitelisted, nil, nil, nil
+	}
+
+	releaseHandshake, err := s.inboundAdmission.AcquireSource(
+		remoteAddr, false,
+	)
+	if err != nil {
+		return whitelisted, nil, nil, err
+	}
+
+	v2Admission := s.inboundAdmission.BindV2(remoteAddr, false)
+	return whitelisted, releaseHandshake, v2Admission, nil
+}
+
 // inboundPeerConnected is invoked by the connection manager when a new inbound
 // connection is established.  It initializes a new inbound server peer
 // instance, associates it with the connection, and starts a goroutine to wait
 // for disconnection.
 func (s *server) inboundPeerConnected(conn net.Conn) {
 	remoteAddr := conn.RemoteAddr()
-	whitelisted := isWhitelisted(remoteAddr)
-
-	// Loopback includes onion peers forwarded into the listener. Bypass the
-	// per-source limits for these and configured whitelisted peers, while the
-	// global socket, v2 rate, and v2 concurrency limits remain active.
-	bypassSourceLimits := whitelisted || inbound.IsLoopback(remoteAddr)
-
-	var releaseHandshake func()
-	if s.inboundAdmission != nil {
-		var err error
-		releaseHandshake, err = s.inboundAdmission.AcquireSource(
-			remoteAddr, bypassSourceLimits,
-		)
-		if err != nil {
-			_ = conn.Close()
-			return
-		}
+	whitelisted, releaseHandshake, v2Admission, err :=
+		s.acquireInboundPeerAdmission(remoteAddr)
+	if err != nil {
+		_ = conn.Close()
+		return
 	}
 
 	sp := newServerPeer(s, false)
@@ -2336,11 +2364,7 @@ func (s *server) inboundPeerConnected(conn net.Conn) {
 	sp.releaseInboundHandshake = releaseHandshake
 
 	peerCfg := newPeerConfig(sp)
-	if s.inboundAdmission != nil {
-		peerCfg.V2HandshakeAdmission = s.inboundAdmission.BindV2(
-			remoteAddr, bypassSourceLimits,
-		)
-	}
+	peerCfg.V2HandshakeAdmission = v2Admission
 
 	sp.Peer = peer.NewInboundPeer(peerCfg)
 	sp.AssociateConnection(conn)
@@ -3205,10 +3229,18 @@ func newServer(listenAddrs, agentBlacklist, agentWhitelist []string,
 	if cfg.MaxPeers < targetOutbound {
 		targetOutbound = cfg.MaxPeers
 	}
-	maxInbound := maxInboundPeers(cfg.MaxPeers, targetOutbound)
+	permanentPeerCount := len(cfg.ConnectPeers)
+	if permanentPeerCount == 0 {
+		permanentPeerCount = len(cfg.AddPeers)
+	}
+	reservedOutbound := reservedOutboundPeers(
+		cfg.MaxPeers, targetOutbound, permanentPeerCount,
+		newAddressFunc != nil,
+	)
+	maxInbound := maxInboundPeers(cfg.MaxPeers, reservedOutbound)
 	if maxInbound == 0 && len(listeners) > 0 {
 		srvrLog.Infof("Inbound connections disabled: maxpeers=%d, "+
-			"reserved-outbound=%d", cfg.MaxPeers, targetOutbound)
+			"reserved-outbound=%d", cfg.MaxPeers, reservedOutbound)
 	}
 	cmgr, err := connmgr.New(&connmgr.Config{
 		Listeners:      listeners,

server_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"net"
 	"os"
 	"path/filepath"
 	"sync/atomic"
@@ -8,6 +9,7 @@ import (
 	"time"
 
 	"github.com/btcsuite/btcd/chaincfg/v2"
+	"github.com/btcsuite/btcd/internal/inbound"
 	"github.com/btcsuite/btcd/peer"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -107,33 +109,165 @@ func TestHandshakeReleaseOnDisconnect(t *testing.T) {
 	require.Equal(t, uint32(1), releases.Load())
 }
 
-// TestMaxInboundPeers verifies that automatic outbound capacity is reserved
-// without underflow at small peer limits.
-func TestMaxInboundPeers(t *testing.T) {
+// TestInboundPeerReservation verifies that listener capacity is derived from
+// the configured peer mode while connmgr retains its automatic target.
+func TestInboundPeerReservation(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
 		name           string
 		maxPeers       int
 		targetOutbound int
-		want           uint32
+		permanentPeers int
+		automatic      bool
+		wantReserved   int
+		wantInbound    uint32
 	}{
-		{name: "zero", maxPeers: 0, targetOutbound: 0, want: 0},
-		{name: "all outbound", maxPeers: 8, targetOutbound: 8, want: 0},
-		{name: "below outbound", maxPeers: 7, targetOutbound: 8, want: 0},
-		{name: "one inbound", maxPeers: 9, targetOutbound: 8, want: 1},
-		{name: "default", maxPeers: 125, targetOutbound: 8, want: 117},
+		{
+			name: "connect only", maxPeers: 8, targetOutbound: 8,
+			permanentPeers: 1, wantReserved: 1, wantInbound: 7,
+		},
+		{
+			name: "connect only capped", maxPeers: 8,
+			targetOutbound: 8, permanentPeers: 10,
+			wantReserved: 8, wantInbound: 0,
+		},
+		{
+			name: "simnet without peers", maxPeers: 8,
+			targetOutbound: 8, wantReserved: 0, wantInbound: 8,
+		},
+		{
+			name: "simnet with peers", maxPeers: 8,
+			targetOutbound: 8, permanentPeers: 3,
+			wantReserved: 3, wantInbound: 5,
+		},
+		{
+			name: "automatic without add peers", maxPeers: 125,
+			targetOutbound: 8, automatic: true,
+			wantReserved: 8, wantInbound: 117,
+		},
+		{
+			name: "add peers below target", maxPeers: 125,
+			targetOutbound: 8, permanentPeers: 3, automatic: true,
+			wantReserved: 11, wantInbound: 114,
+		},
+		{
+			name: "add peers above target", maxPeers: 10,
+			targetOutbound: 8, permanentPeers: 9, automatic: true,
+			wantReserved: 10, wantInbound: 0,
+		},
+		{
+			name: "add peers at max peers", maxPeers: 10,
+			targetOutbound: 8, permanentPeers: 10, automatic: true,
+			wantReserved: 10, wantInbound: 0,
+		},
+		{
+			name: "add peers above max peers", maxPeers: 10,
+			targetOutbound: 8, permanentPeers: 12, automatic: true,
+			wantReserved: 10, wantInbound: 0,
+		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			require.Equal(t, test.want, maxInboundPeers(
+			reserved := reservedOutboundPeers(
 				test.maxPeers, test.targetOutbound,
+				test.permanentPeers, test.automatic,
+			)
+			require.Equal(t, test.wantReserved, reserved)
+			require.Equal(t, test.wantInbound, maxInboundPeers(
+				test.maxPeers, reserved,
 			))
 		})
 	}
 }
 
+// TestInboundPeerAdmissionSourceLimits verifies that loopback and whitelisted
+// peers retain the ordinary pending-handshake and V2 source limits.
+func TestInboundPeerAdmissionSourceLimits(t *testing.T) {
+	_, whitelist, err := net.ParseCIDR("192.0.2.0/24")
+	require.NoError(t, err)
+
+	originalCfg := cfg
+	t.Cleanup(func() {
+		cfg = originalCfg
+	})
+
+	tests := []struct {
+		name            string
+		addr            net.Addr
+		whitelists      []*net.IPNet
+		wantWhitelisted bool
+	}{
+		{
+			name: "loopback",
+			addr: &net.TCPAddr{
+				IP: net.ParseIP("127.0.0.2"), Port: 8333,
+			},
+		},
+		{
+			name: "whitelisted",
+			addr: &net.TCPAddr{
+				IP: net.ParseIP("192.0.2.1"), Port: 8333,
+			},
+			whitelists:      []*net.IPNet{whitelist},
+			wantWhitelisted: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			cfg = &config{whitelists: test.whitelists}
+			s := &server{inboundAdmission: inbound.New()}
+
+			var releases []func()
+			for i := 0; i < 20; i++ {
+				whitelisted, release, _, err :=
+					s.acquireInboundPeerAdmission(test.addr)
+				if err != nil {
+					break
+				}
+
+				require.Equal(
+					t, test.wantWhitelisted, whitelisted,
+				)
+				releases = append(releases, release)
+			}
+			require.Less(t, len(releases), 20,
+				"the source pending limit must reject a peer")
+			for _, release := range releases {
+				release()
+			}
+
+			var v2Rejections int
+			for i := 0; i < 20; i++ {
+				whitelisted, releaseSource, v2Admission, err :=
+					s.acquireInboundPeerAdmission(test.addr)
+				require.NoError(t, err)
+				require.Equal(
+					t, test.wantWhitelisted, whitelisted,
+				)
+				releaseSource()
+
+				releaseV2, err := v2Admission.Acquire()
+				if err != nil {
+					v2Rejections++
+					break
+				}
+				releaseV2()
+
+				releaseV2, err = v2Admission.Acquire()
+				require.NoError(t, err,
+					"the second CPU phase must not consume "+
+						"another rate token")
+				releaseV2()
+			}
+			require.Equal(t, 1, v2Rejections,
+				"the V2 source rate must reject a peer")
+		})
+	}
+}
+
 // TestPeerLifecycleOrdering verifies that when verack arrives before
 // disconnect, peerLifecycleHandler emits peerAdd followed by peerDone
 // on the peerLifecycle channel -- never out of order.