feat(socket): enforce SO_RCVBUF on the raw receive queue

Extends the UDP SO_RCVBUF enforcement to RawSocket for symmetry:
process_raw_packet drops an inbound packet whose payload would push the
queued receive bytes past the cap (Linux sk_rcvqueues_full), enforced
only when SO_RCVBUF is set; unset stays unbounded.

Tests-first: two unit tests in test__runtime__socket__raw__socket.py
(over-cap drop, unset-is-unbounded).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: ccie18643

packages/pytcp/pytcp/runtime/socket/raw__socket.py

@@ -544,6 +544,19 @@ def process_raw_packet(self, packet_rx_md: RawMetadata) -> None:
         with self._lock__io:
             if self._closed:
                 return
+            # SO_RCVBUF enforcement (Linux 'sk_rcvqueues_full'): once the
+            # operator sets a receive-buffer cap, drop an inbound packet
+            # whose payload would push the queued bytes past it. Unset
+            # ('None') stays unbounded. Measured against summed payload
+            # bytes (the Linux 'truesize' overhead is not modelled).
+            if self._so_rcvbuf is not None:
+                queued = sum(len(md.raw__data) for md in self._packet_rx_md)
+                if queued + len(packet_rx_md.raw__data) > self._so_rcvbuf:
+                    __debug__ and log(
+                        "socket",
+                        f"<g>[{self}]</> - Dropped packet: SO_RCVBUF cap " f"{self._so_rcvbuf} exceeded",
+                    )
+                    return
             self._packet_rx_md.append(packet_rx_md)
             self._packet_rx_md_ready.release()
         self._signal_readable()

packages/pytcp/pytcp/tests/unit/runtime/socket/test__runtime__socket__raw__socket.py

@@ -52,6 +52,8 @@
     IPPROTO_IPV6,
     IPV6_HOPLIMIT,
     IPV6_RECVHOPLIMIT,
+    SO_RCVBUF,
+    SOL_SOCKET,
     AddressFamily,
     SocketType,
     gaierror,
@@ -1354,3 +1356,60 @@ def test__raw_socket__non_icmp6_payload_unchanged(self) -> None:
         )
 
         self.assertEqual(result, payload, msg="A non-ICMPv6 raw socket must not rewrite the payload.")
+
+
+class TestRawSocketRcvbuf(_RawSocketTestCase):
+    """
+    The 'RawSocket' SO_RCVBUF receive-queue-cap tests.
+    """
+
+    def _make_md(self) -> RawMetadata:
+        """
+        Build a canonical IPv4 'RawMetadata' envelope with a 7-byte
+        payload.
+        """
+
+        return RawMetadata(
+            ip__ver=IpVersion.IP4,
+            ip__local_address=Ip4Address("10.0.0.1"),
+            ip__remote_address=Ip4Address("10.0.0.2"),
+            ip__proto=IpProto.ICMP4,
+            raw__data=b"payload",
+        )
+
+    def test__raw_socket__so_rcvbuf_drops_packet_over_cap(self) -> None:
+        """
+        Ensure that once 'SO_RCVBUF' is set, an inbound packet whose
+        payload would push the queued receive bytes past the cap is
+        dropped rather than enqueued.
+
+        Reference: RFC 1122 §4.2.2.16 (receive-buffer bound).
+        """
+
+        s = RawSocket(family=AddressFamily.INET4, protocol=IpProto.ICMP4)
+        s.setsockopt(SOL_SOCKET, SO_RCVBUF, 20)
+        s.process_raw_packet(self._make_md())
+        s.process_raw_packet(self._make_md())
+        s.process_raw_packet(self._make_md())
+        self.assertEqual(
+            len(s._packet_rx_md),
+            2,
+            msg="A raw packet over the SO_RCVBUF cap must be dropped, not enqueued.",
+        )
+
+    def test__raw_socket__so_rcvbuf_unset_is_unbounded(self) -> None:
+        """
+        Ensure that with 'SO_RCVBUF' unset the raw receive queue is
+        unbounded (no default cap).
+
+        Reference: PyTCP test infrastructure (no RFC clause).
+        """
+
+        s = RawSocket(family=AddressFamily.INET4, protocol=IpProto.ICMP4)
+        for _ in range(50):
+            s.process_raw_packet(self._make_md())
+        self.assertEqual(
+            len(s._packet_rx_md),
+            50,
+            msg="With SO_RCVBUF unset every raw packet must be enqueued.",
+        )