diff --git a/config.go b/config.go index 3178557a..a15579b0 100644 --- a/config.go +++ b/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"` diff --git a/internal/inbound/admission.go b/internal/inbound/admission.go index acfe68db..0d72f9fd 100644 --- a/internal/inbound/admission.go +++ b/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 } diff --git a/internal/inbound/admission_test.go b/internal/inbound/admission_test.go index a29915b4..8bead975 100644 --- a/internal/inbound/admission_test.go +++ b/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) { diff --git a/peer/peer_test.go b/peer/peer_test.go index e7a66584..733029ec 100644 --- a/peer/peer_test.go +++ b/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) } } diff --git a/server.go b/server.go index a60875db..678e3f82 100644 --- a/server.go +++ b/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, diff --git a/server_test.go b/server_test.go index b205e966..5639b595 100644 --- a/server_test.go +++ b/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.