refactor(drain): collapse the four socket drains into one

[benaliabderrahmane]

Description

The receive loop existed in four near-identical copies: drain_subscription
in rmw_subscription.cpp, drain_socket in rmw_wait.cpp, and an inline copy in
each of rmw_take_request and rmw_take_response. They had drifted, and the
drift is where the behavior gaps live — each copy decided for itself whether to
cap the queue, warn about an overflow, or notify a listener, so the same message
was treated differently depending on which path delivered it.

drain_endpoint() is now the only receive loop. DrainTarget carries the
per-endpoint policy (queue and its mutex, depth cap, accepted msg_type, shm
reader cache, ignore_local_publications, the TRANSIENT_LOCAL watermark map, the
listener callback, the name to log) and the drain_target() overloads build it
from each endpoint's impl struct, so a call site can no longer pick its own
policy by accident.

This is the riskiest change in the series — it rewrites the hot receive path.
It is deliberately a standalone PR, with no feature on top, so the diff can be
read against the four originals.

Is this user-facing behavior change?

Yes — three differences that could not be preserved once the four copies became
one. All three are the copies disagreeing, resolved in favour of the stricter one:

• The rmw_wait drain now fires a subscription's on_new_message callback.
  Only the rmw_take path did, so a callback registered by an executor was
  silent for every message that arrived while a thread sat in rmw_wait.
• The rmw_wait drain now reports queue overflow. It trimmed to the QoS depth
  silently, so messages dropped by a slow subscriber were only visible when the
  drop happened to land on a take.
• Request and response queues are capped on the take path too. rmw_wait
  has always trimmed both to 100; the take-path copies had no cap at all, so
  a caller driving a service with rmw_take_request() alone and falling behind
  grew the queue without limit. That bound is now named SERVICE_QUEUE_DEPTH
  and applied wherever the queue is filled.

Everything else is preserved exactly: check order (msg_type, replay dedup,
descriptor resolve, ignore_local), the QoS depth cap on subscriptions, the
per-message callback count of 1, and the receive timestamp.

How was this tested?

No new tests — the existing suite is the safety net for a refactor, and
test_rmw_qos.cpp (1572 lines, 20 TRANSIENT_LOCAL tests) is the one that
matters here: the unified drain has to preserve the replayed_watermarks dedup
and the depth trim. test_rmw_wait.cpp covers the lost-wakeup regressions in the
file this touches.

Clean rebuild (not incremental) plus full suite on jazzy: 156 tests, 0
failures. kilted, rolling and lyrical are on CI.

Did you use Generative AI?

Additional Information

Independent of #1 in this series; PRs 3-6 build on this one, so it wants to land
first among those.

Incidental finding while merging the copies: the five clock helpers involved
(system_now_ns in four files, wall_now_ns in rmw_wait.cpp) were all the
same system_clock nanosecond reading under different names, so consolidating
them to one is a rename, not a semantic change.