Control-plane sub/unsub clone in stream loop

[muscariello]

File: core/datapath/src/message_processing.rs — process_stream() (L985)

Problem:

let _ = txcp.send(Ok(msg.clone())).await;

Every remote subscribe/unsubscribe message is cloned before forwarding to the
control-plane channel, even though the original msg is consumed immediately
after.

Mitigation plan:
Send the original message to the control plane and reconstruct (or skip) the
local processing copy, or clone only the small header fields needed by the
control plane.