peer: enforce onion message rate limits at ingress

diff --git a/lnwire/onion_message.go b/lnwire/onion_message.go
index 5d56808..a9f13d8 100644
--- a/lnwire/onion_message.go
+++ b/lnwire/onion_message.go
@@ -82,9 +82,26 @@ func (o *OnionMessage) MsgType() MessageType {
 	return MsgOnionMessage
 }
 
+// WireSize returns the on-the-wire size of the message in bytes, including
+// the 2-byte message type prefix, the 33-byte compressed path key, the
+// 2-byte onion blob length prefix, and the onion blob itself. It computes
+// the size directly from the in-memory fields rather than round-tripping
+// through Encode, so callers in the hot ingress path — notably the onion
+// message rate limiter — can charge the right number of byte tokens
+// without paying for a full serialization.
+func (o *OnionMessage) WireSize() int {
+	const (
+		msgTypeBytes  = 2
+		pathKeyBytes  = 33
+		onionLenBytes = 2
+	)
+
+	return msgTypeBytes + pathKeyBytes + onionLenBytes + len(o.OnionBlob)
+}
+
 // SerializedSize returns the serialized size of the message in bytes.
 //
 // This is part of the lnwire.SizeableMessage interface.
 func (o *OnionMessage) SerializedSize() (uint32, error) {
-	return MessageSerializedSize(o)
+	return uint32(o.WireSize()), nil
 }
diff --git a/lnwire/onion_message_test.go b/lnwire/onion_message_test.go
new file mode 100644
index 0000000..bd01e33
--- /dev/null
+++ b/lnwire/onion_message_test.go
@@ -0,0 +1,39 @@
+package lnwire
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"pgregory.net/rapid"
+)
+
+// TestOnionMessageWireSizeMatchesEncode verifies that the value produced by
+// OnionMessage.WireSize — the value fed to the onion message rate limiter on
+// every incoming packet — matches the number of bytes WriteMessage actually
+// emits for that same message. WireSize computes its result directly from the
+// in-memory fields without round-tripping through Encode, which is fast but
+// creates a risk of silent divergence if the OnionMessage wire format ever
+// gains an optional TLV extension or extra field. This test is the
+// compile-time-cheap regression guard that divergence does not go undetected.
+func TestOnionMessageWireSizeMatchesEncode(t *testing.T) {
+	t.Parallel()
+
+	rapid.Check(t, func(rt *rapid.T) {
+		msg, ok := (*OnionMessage)(nil).RandTestMessage(
+			rt,
+		).(*OnionMessage)
+		require.True(
+			rt, ok, "RandTestMessage did "+
+				"not return an OnionMessage",
+		)
+
+		var buf bytes.Buffer
+		written, err := WriteMessage(&buf, msg, 0)
+		require.NoError(rt, err, "WriteMessage error")
+		require.Equal(rt, written, msg.WireSize(),
+			"WireSize=%d, WriteMessage wrote=%d bytes",
+			msg.WireSize(), written,
+		)
+	})
+}
diff --git a/peer/brontide.go b/peer/brontide.go
index 2ec32b0..a946273 100644
--- a/peer/brontide.go
+++ b/peer/brontide.go
@@ -315,6 +315,19 @@ type Config struct {
 	// message actor. If nil, onion messaging is disabled.
 	SpawnOnionActor onionmessage.OnionActorFactory
 
+	// OnionLimiter is the combined per-peer + global onion message
+	// ingress rate limiter. It hides the split between the two
+	// underlying buckets behind a single interface: callers invoke
+	// OnionLimiter.AllowN on every incoming onion message and it
+	// consults the per-peer bucket first (so a hostile peer whose own
+	// budget is empty cannot drain the shared budget on rejected
+	// attempts) and then the global bucket. Per-peer state is retained
+	// across disconnect so a peer cannot reset its bucket by cycling
+	// the connection; see the PeerRateLimiter doc for the memory-bound
+	// argument. A nil value means onion message rate limiting is
+	// disabled.
+	OnionLimiter onionmessage.IngressLimiter
+
 	// OnionActorOpts returns ActorOptions for the onion peer actor
 	// being spawned for the given peer. This allows per-peer
 	// customization of mailbox size, drop predicates, etc.
@@ -2331,6 +2344,32 @@ out:
 			discStream.AddMsg(msg)
 
 		case *lnwire.OnionMessage:
+			// Charge the limiter the on-the-wire size of the
+			// message so the byte-granular bucket reflects
+			// actual ingress bandwidth rather than raw message
+			// counts. A rejection surfaces as a sentinel error
+			// wrapped in fn.Result; errors.Is lets us pick the
+			// right first-drop log path.
+			result := allowOnionMessage(
+				p.cfg.OnionLimiter, p.PubKey(),
+				msg.WireSize(),
+			)
+			if err := result.Err(); err != nil {
+				logFirstOnionDrop(
+					peerLog, p.log, err,
+					p.cfg.OnionLimiter,
+				)
+				// Keep repeated drops at trace so a
+				// sustained attack does not flood debug;
+				// the first-drop info log above already
+				// gives operators a clear "limiter
+				// engaged" signal.
+				p.log.Tracef("dropping onion message: %v",
+					err)
+
+				break
+			}
+
 			p.onionActorRef.WhenSome(
 				func(ref onionmessage.OnionPeerActorRef) {
 					// TODO(elle): thread contexts through
diff --git a/peer/onion_ratelimit.go b/peer/onion_ratelimit.go
new file mode 100644
index 0000000..b9c9dcc
--- /dev/null
+++ b/peer/onion_ratelimit.go
@@ -0,0 +1,56 @@
+package peer
+
+import (
+	"errors"
+
+	"github.com/btcsuite/btclog/v2"
+	"github.com/lightningnetwork/lnd/fn/v2"
+	"github.com/lightningnetwork/lnd/onionmessage"
+)
+
+// allowOnionMessage delegates to the IngressLimiter for the
+// per-peer-then-global byte-granular rate limit check. A successful
+// result wraps fn.Unit; a rejection wraps one of the sentinel errors
+// onionmessage.ErrPeerRateLimit or onionmessage.ErrGlobalRateLimit so
+// that callers can distinguish the drop reason via errors.Is.
+//
+// A nil IngressLimiter is treated as "disabled" and always accepts the
+// message. This preserves the behavior of test and disabled-onion-
+// messaging configurations without forcing callers to construct a real
+// limiter.
+func allowOnionMessage(limiter onionmessage.IngressLimiter,
+	peerKey [33]byte, msgBytes int) fn.Result[fn.Unit] {
+
+	if limiter == nil {
+		return fn.Ok(fn.Unit{})
+	}
+
+	return limiter.AllowN(peerKey, msgBytes)
+}
+
+// logFirstOnionDrop emits a one-shot info log the first time the limiter
+// identified by err trips. Per-peer drops go to peerLog (caller's
+// peer-prefixed log) so the operator can see which peer first tripped
+// the limiter; global drops go to pkgLog (typically the package-level
+// peerLog) since they are not attributable to any single peer.
+func logFirstOnionDrop(pkgLog, peerLog btclog.Logger, err error,
+	limiter onionmessage.IngressLimiter) {
+
+	if limiter == nil {
+		return
+	}
+
+	switch {
+	case errors.Is(err, onionmessage.ErrGlobalRateLimit):
+		if limiter.FirstGlobalDropClaim() {
+			pkgLog.Infof("onion message global rate limiter " +
+				"engaged; further drops logged at trace")
+		}
+
+	case errors.Is(err, onionmessage.ErrPeerRateLimit):
+		if limiter.FirstPeerDropClaim() {
+			peerLog.Infof("onion message per-peer rate limiter " +
+				"engaged; further drops logged at trace")
+		}
+	}
+}
diff --git a/peer/onion_ratelimit_log_test.go b/peer/onion_ratelimit_log_test.go
new file mode 100644
index 0000000..ffac40d
--- /dev/null
+++ b/peer/onion_ratelimit_log_test.go
@@ -0,0 +1,123 @@
+package peer
+
+import (
+	"bytes"
+	"errors"
+	"testing"
+
+	"github.com/btcsuite/btclog/v2"
+	"github.com/lightningnetwork/lnd/onionmessage"
+	"github.com/stretchr/testify/require"
+)
+
+// newCapturingLogger builds a btclog.Logger backed by an in-memory buffer
+// so tests can assert whether a given log line was emitted.
+func newCapturingLogger() (btclog.Logger, *bytes.Buffer) {
+	buf := &bytes.Buffer{}
+	handler := btclog.NewDefaultHandler(buf, btclog.WithNoTimestamp())
+	return btclog.NewSLogger(handler), buf
+}
+
+// newRealIngressLimiter constructs a real ingressLimiter backed by real
+// per-peer and global limiters sized so the first message passes and
+// every subsequent one trips the named side of the limiter. It is used
+// by the log tests to exercise the one-shot claim path against real
+// FirstDropClaim bookkeeping rather than a stub.
+func newRealIngressLimiter(t *testing.T) onionmessage.IngressLimiter {
+	t.Helper()
+
+	// Burst == one max-sized message for both sides; rate of 1 Kbps
+	// ensures neither bucket refills within the test window.
+	peerLim := onionmessage.NewPeerRateLimiter(1, testMsgBytes)
+	globalLim := onionmessage.NewGlobalLimiter(1, testMsgBytes)
+
+	return onionmessage.NewIngressLimiter(peerLim, globalLim)
+}
+
+// TestLogFirstOnionDropGlobalOneShot verifies that logFirstOnionDrop
+// emits exactly one info-level line for the global limiter's first
+// drop and is silent on subsequent drops, so operators get a single
+// "engaged" signal without log flooding under sustained attack. The
+// global first-drop line must land on the package-level logger, not
+// the per-peer one, since a global drop is not attributable to any
+// single peer.
+func TestLogFirstOnionDropGlobalOneShot(t *testing.T) {
+	t.Parallel()
+
+	pkgLog, pkgBuf := newCapturingLogger()
+	peerLog, peerBuf := newCapturingLogger()
+	limiter := newRealIngressLimiter(t)
+
+	// First drop log: must emit to the package-level logger.
+	logFirstOnionDrop(
+		pkgLog, peerLog, onionmessage.ErrGlobalRateLimit, limiter,
+	)
+	require.Contains(t, pkgBuf.String(), "global rate limiter")
+	require.Empty(t, peerBuf.String(),
+		"global drop must not land on the peer-prefix log")
+
+	// Second drop log: must be silent (both buffer sizes unchanged).
+	sizeAfterFirst := pkgBuf.Len()
+	logFirstOnionDrop(
+		pkgLog, peerLog, onionmessage.ErrGlobalRateLimit, limiter,
+	)
+	require.Equal(t, sizeAfterFirst, pkgBuf.Len(),
+		"second drop must not re-log the first-drop line")
+	require.Empty(t, peerBuf.String())
+}
+
+// TestLogFirstOnionDropPeerOneShot verifies the same one-shot property
+// for the per-peer limiter and that the nil-limiter guard prevents a
+// panic when onion message rate limiting is entirely disabled. The
+// per-peer first-drop line must land on the peer-prefix logger so
+// operators can see which peer tripped the limiter.
+func TestLogFirstOnionDropPeerOneShot(t *testing.T) {
+	t.Parallel()
+
+	pkgLog, pkgBuf := newCapturingLogger()
+	peerLog, peerBuf := newCapturingLogger()
+
+	// Nil limiter: must not panic and must not log to either logger.
+	logFirstOnionDrop(
+		pkgLog, peerLog, onionmessage.ErrPeerRateLimit, nil,
+	)
+	require.Empty(t, pkgBuf.String())
+	require.Empty(t, peerBuf.String())
+
+	// Real limiter: emit once to the peer logger, then silent.
+	limiter := newRealIngressLimiter(t)
+	logFirstOnionDrop(
+		pkgLog, peerLog, onionmessage.ErrPeerRateLimit, limiter,
+	)
+	require.Contains(t, peerBuf.String(), "per-peer rate limiter")
+	require.Empty(t, pkgBuf.String(),
+		"per-peer drop must not land on the package-level log")
+	sizeAfterFirst := peerBuf.Len()
+	logFirstOnionDrop(
+		pkgLog, peerLog, onionmessage.ErrPeerRateLimit, limiter,
+	)
+	require.Equal(t, sizeAfterFirst, peerBuf.Len())
+}
+
+// TestLogFirstOnionDropUnknownReason verifies that an error that does
+// not match any known drop reason is a no-op — neither limiter's
+// first-drop flag is consumed. This guards against a typo or a new
+// drop reason being added without a matching log case.
+func TestLogFirstOnionDropUnknownReason(t *testing.T) {
+	t.Parallel()
+
+	pkgLog, pkgBuf := newCapturingLogger()
+	peerLog, peerBuf := newCapturingLogger()
+	limiter := newRealIngressLimiter(t)
+
+	logFirstOnionDrop(pkgLog, peerLog, errors.New("unknown"), limiter)
+	require.Empty(t, pkgBuf.String())
+	require.Empty(t, peerBuf.String())
+
+	// Both limiters' first-drop flags must still be unclaimed, so a
+	// follow-up call with a valid reason still emits the info line.
+	logFirstOnionDrop(
+		pkgLog, peerLog, onionmessage.ErrGlobalRateLimit, limiter,
+	)
+	require.Contains(t, pkgBuf.String(), "global rate limiter")
+}
diff --git a/peer/onion_ratelimit_test.go b/peer/onion_ratelimit_test.go
new file mode 100644
index 0000000..af8031c
--- /dev/null
+++ b/peer/onion_ratelimit_test.go
@@ -0,0 +1,263 @@
+package peer
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/lightningnetwork/lnd/fn/v2"
+	"github.com/lightningnetwork/lnd/onionmessage"
+	"github.com/stretchr/testify/require"
+)
+
+// testMsgBytes is the on-the-wire size we charge the bucket per call in
+// these tests. It is sized to approximate a spec-max onion message so
+// that burst budgets scale naturally with the per-message cost.
+const testMsgBytes = 32 * 1024
+
+// stubIngressLimiter is a test double for onionmessage.IngressLimiter
+// that records every call and delegates the accept/reject decision to a
+// caller-supplied predicate.
+type stubIngressLimiter struct {
+	// decide is invoked for every AllowN call. It receives the peer
+	// key and byte count and returns the error to embed in the
+	// fn.Result — nil for accept.
+	decide func(peer [33]byte, n int) error
+
+	calls atomic.Uint64
+}
+
+// AllowN records the call and dispatches to the configured predicate.
+func (s *stubIngressLimiter) AllowN(peer [33]byte,
+	n int) fn.Result[fn.Unit] {
+
+	s.calls.Add(1)
+	if err := s.decide(peer, n); err != nil {
+		return fn.Err[fn.Unit](err)
+	}
+
+	return fn.Ok(fn.Unit{})
+}
+
+// FirstPeerDropClaim always returns true so the log-path test can
+// observe the one-shot dispatch. Tests that care about the one-shot
+// invariant use a real IngressLimiter instead.
+func (s *stubIngressLimiter) FirstPeerDropClaim() bool { return true }
+
+// FirstGlobalDropClaim always returns true for the same reason.
+func (s *stubIngressLimiter) FirstGlobalDropClaim() bool { return true }
+
+// TestAllowOnionMessageNilLimiter verifies that allowOnionMessage treats
+// a nil IngressLimiter as "disabled" and unconditionally accepts
+// messages.
+func TestAllowOnionMessageNilLimiter(t *testing.T) {
+	t.Parallel()
+
+	var peer [33]byte
+	result := allowOnionMessage(nil, peer, testMsgBytes)
+	require.NoError(t, result.Err())
+}
+
+// TestAllowOnionMessagePeerRejectsFirst verifies that a real
+// IngressLimiter consults the per-peer limiter before the global
+// limiter: once the per-peer bucket is drained, the global bucket
+// must not be touched on subsequent calls, preserving the shared
+// budget against a hostile peer burning global tokens via rejected
+// attempts.
+func TestAllowOnionMessagePeerRejectsFirst(t *testing.T) {
+	t.Parallel()
+
+	// Real per-peer limiter with burst of exactly one message; very
+	// low rate so it does not refill during the test.
+	peerLim := onionmessage.NewPeerRateLimiter(1, testMsgBytes)
+
+	// Stub "global" that records whether it was consulted. It wraps
+	// the global side of the IngressLimiter.
+	globalCalls := atomic.Uint64{}
+	global := &countingGlobalStub{
+		allow: func() bool { return true },
+		calls: &globalCalls,
+	}
+
+	limiter := onionmessage.NewIngressLimiter(peerLim, global)
+
+	var key [33]byte
+	key[0] = 0x03
+
+	// First call drains the per-peer bucket; both limiters are
+	// consulted so global.calls bumps to 1.
+	result := allowOnionMessage(limiter, key, testMsgBytes)
+	require.NoError(t, result.Err())
+	require.Equal(t, uint64(1), globalCalls.Load())
+
+	// Second call trips the per-peer limiter and must NOT consult
+	// the global limiter — globalCalls stays at 1.
+	result = allowOnionMessage(limiter, key, testMsgBytes)
+	require.Error(t, result.Err())
+	require.True(t,
+		errors.Is(result.Err(), onionmessage.ErrPeerRateLimit),
+	)
+	require.Equal(t, uint64(1), peerLim.Dropped())
+	require.Equal(t, uint64(1), globalCalls.Load(),
+		"global limiter must not be consulted when per-peer rejects")
+}
+
+// countingGlobalStub is a minimal RateLimiter test double that counts
+// calls to AllowN and delegates the accept/reject decision to a
+// caller-supplied predicate. It exists so tests can feed a real
+// ingressLimiter a controllable global side.
+type countingGlobalStub struct {
+	allow func() bool
+	calls *atomic.Uint64
+}
+
+func (s *countingGlobalStub) AllowN(_ int) bool {
+	s.calls.Add(1)
+
+	return s.allow()
+}
+
+// TestAllowOnionMessageGlobalRejects verifies that when the per-peer
+// limiter permits traffic but the global bucket is exhausted,
+// allowOnionMessage surfaces ErrGlobalRateLimit.
+func TestAllowOnionMessageGlobalRejects(t *testing.T) {
+	t.Parallel()
+
+	peerLim := onionmessage.NewPeerRateLimiter(
+		1_000_000, 100*testMsgBytes,
+	)
+
+	globalCalls := atomic.Uint64{}
+	global := &countingGlobalStub{
+		allow: func() bool { return false },
+		calls: &globalCalls,
+	}
+	limiter := onionmessage.NewIngressLimiter(peerLim, global)
+
+	var key [33]byte
+	key[0] = 0x02
+
+	result := allowOnionMessage(limiter, key, testMsgBytes)
+	require.Error(t, result.Err())
+	require.True(t,
+		errors.Is(result.Err(), onionmessage.ErrGlobalRateLimit),
+	)
+	require.Equal(t, uint64(0), peerLim.Dropped())
+	require.Equal(t, uint64(1), globalCalls.Load())
+}
+
+// TestAllowOnionMessageHappyPath verifies that a fully-configured
+// IngressLimiter accepts a stream of messages when neither bucket is
+// under pressure.
+func TestAllowOnionMessageHappyPath(t *testing.T) {
+	t.Parallel()
+
+	peerLim := onionmessage.NewPeerRateLimiter(
+		1_000_000, 100*testMsgBytes,
+	)
+	globalCalls := atomic.Uint64{}
+	global := &countingGlobalStub{
+		allow: func() bool { return true },
+		calls: &globalCalls,
+	}
+	limiter := onionmessage.NewIngressLimiter(peerLim, global)
+
+	var key [33]byte
+	key[0] = 0x04
+
+	for i := 0; i < 10; i++ {
+		result := allowOnionMessage(limiter, key, testMsgBytes)
+		require.NoError(t, result.Err(), "iter %d", i)
+	}
+	require.Equal(t, uint64(0), peerLim.Dropped())
+}
+
+// TestAllowOnionMessagePeerIsolation verifies at the peer-package level
+// that exhausting one peer's bucket through allowOnionMessage does not
+// affect a different peer's allowance — guarding against a regression
+// where the helper might key the bucket incorrectly.
+func TestAllowOnionMessagePeerIsolation(t *testing.T) {
+	t.Parallel()
+
+	peerLim := onionmessage.NewPeerRateLimiter(1, 2*testMsgBytes)
+	globalCalls := atomic.Uint64{}
+	global := &countingGlobalStub{
+		allow: func() bool { return true },
+		calls: &globalCalls,
+	}
+	limiter := onionmessage.NewIngressLimiter(peerLim, global)
+
+	var keyA, keyB [33]byte
+	keyA[0] = 0x02
+	keyB[0] = 0x03
+
+	// Drain peer A.
+	for i := 0; i < 2; i++ {
+		result := allowOnionMessage(limiter, keyA, testMsgBytes)
+		require.NoError(t, result.Err())
+	}
+	result := allowOnionMessage(limiter, keyA, testMsgBytes)
+	require.Error(t, result.Err())
+
+	// Peer B must still have its full burst available.
+	for i := 0; i < 2; i++ {
+		result := allowOnionMessage(limiter, keyB, testMsgBytes)
+		require.NoError(t, result.Err(), "peer B slot %d", i)
+	}
+}
+
+// TestAllowOnionMessageConcurrent exercises concurrent access to
+// allowOnionMessage across many goroutines. It asserts that the sum of
+// accepted calls plus the per-peer dropped counter equals the total
+// number of attempts, and that no race or panic occurs. Run with -race
+// for the strongest signal.
+func TestAllowOnionMessageConcurrent(t *testing.T) {
+	t.Parallel()
+
+	const burstMessages = 32
+	peerLim := onionmessage.NewPeerRateLimiter(
+		1, burstMessages*testMsgBytes,
+	)
+	globalCalls := atomic.Uint64{}
+	global := &countingGlobalStub{
+		allow: func() bool { return true },
+		calls: &globalCalls,
+	}
+	limiter := onionmessage.NewIngressLimiter(peerLim, global)
+
+	var key [33]byte
+	key[0] = 0x05
+
+	const workers = 16
+	const perWorker = 64
+	var wg sync.WaitGroup
+	var accepted atomic.Uint64
+
+	for w := 0; w < workers; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := 0; i < perWorker; i++ {
+				result := allowOnionMessage(
+					limiter, key, testMsgBytes,
+				)
+				if result.Err() == nil {
+					accepted.Add(1)
+				}
+			}
+		}()
+	}
+	wg.Wait()
+
+	total := uint64(workers * perWorker)
+	require.Equal(
+		t, total, accepted.Load()+peerLim.Dropped(),
+		"every attempt must be counted as accepted or dropped",
+	)
+	// With a near-zero refill rate the bucket can only issue at most
+	// burstMessages accepts before refill; since the test runs much
+	// faster than the refill interval, accepted should equal the
+	// burst.
+	require.Equal(t, uint64(burstMessages), accepted.Load())
+}