fix(uORB): snapshot _data under the node lock before sending to a remote subscriber

process_add_subscription() (CONFIG_ORB_COMMUNICATOR only) read _data_valid and
the topic buffer contents while passing a pointer into _data straight to
send_message(), with no synchronization against a concurrent publisher. A
publisher's write() memcpy's into _data and sets _data_valid under ATOMIC_ENTER,
so the remote-init send can read a torn/partially-updated sample - most clearly
for o_queue == 1, where the read slot ((gen - 1) % 1) is exactly the slot being
written.

This is a pre-existing latent race (the previous px4::atomic<uint8_t*> _data only
guarded the pointer, never the bytes) that neither ASan nor TSan can catch because
CONFIG_ORB_COMMUNICATOR is not built in those configs. On VOXL2 it runs on the
muorb RX thread concurrently with publishers.

Snapshot the most recent sample into a temporary buffer under ATOMIC_ENTER, then
send it outside the critical section - mirroring copy(). The buffer is allocated
before the critical section (it cannot be allocated under ATOMIC_ENTER on NuttX)
and send_message() must run outside it anyway, since it can be slow and may call
back into DeviceNode. This only affects the rare subscription-add path, not the
publish hot path.

Signed-off-by: Julian Oes <julian@oes.ch>

platforms/common/uORB/uORBDeviceNode.cpp

@@ -410,18 +410,43 @@ void uORB::DeviceNode::remove_internal_subscriber()
 #ifdef CONFIG_ORB_COMMUNICATOR
 int16_t uORB::DeviceNode::process_add_subscription()
 {
-	// if there is already data in the node, send this out to
-	// the remote entity.
-	// send the data to the remote entity.
+	// If there is already data in the node, send this out to the remote entity to
+	// initialize its end.
 	uORBCommunicator::IChannel *ch = uORB::Manager::get_instance()->get_uorb_communicator();
 
-	if (_data != nullptr && ch != nullptr) { // _data will not be null if there is a publisher.
-		// Only send the most recent data to initialize the remote end.
-		if (_data_valid) {
-			ch->send_message(_meta->o_name, _meta->o_size, _data + (_meta->o_size * ((_generation.load() - 1) % _meta->o_queue)));
+	if (ch == nullptr) {
+		return PX4_OK;
 	}
+
+	// Snapshot the most recent sample under the node lock, then send it outside the
+	// critical section. Reading _data / _data_valid and the buffer contents while
+	// sending would race with a concurrent write(), which memcpy's into _data and
+	// sets _data_valid under ATOMIC_ENTER. send_message() must also not run under
+	// ATOMIC_ENTER: it can be slow and may call back into DeviceNode. This mirrors the
+	// snapshot-under-lock done in copy().
+	uint8_t *buffer = (uint8_t *)px4_cache_aligned_alloc(_meta->o_size);
+
+	if (buffer == nullptr) {
+		return PX4_ERROR;
 	}
 
+	bool have_data = false;
+
+	ATOMIC_ENTER;
+
+	if (_data != nullptr && _data_valid) { // _data will not be null if there is a publisher.
+		memcpy(buffer, _data + (_meta->o_size * ((_generation.load() - 1) % _meta->o_queue)), _meta->o_size);
+		have_data = true;
+	}
+
+	ATOMIC_LEAVE;
+
+	if (have_data) {
+		ch->send_message(_meta->o_name, _meta->o_size, buffer);
+	}
+
+	free(buffer);
+
 	return PX4_OK;
 }