fix(registry): publish slot state only after payload commits under the seqlock [P1]

rmw_unix_socket_cpp/src/registry.cpp

@@ -170,7 +170,9 @@ static bool snapshot_slot(RegistryEntrySlot * slot, RegistryEntrySlot * out)
       continue;
     }
     uint8_t st = slot->state.load(std::memory_order_acquire);
-    if (st == ENTRY_EMPTY) {
+    if (st == ENTRY_EMPTY || st == ENTRY_RESERVED) {
+      // RESERVED = a writer claimed the slot but has not yet published its
+      // payload. Treat it as empty so we never snapshot a half-built slot.
       return false;
     }
 
@@ -230,12 +232,17 @@ static int32_t try_add_once(RegistryHeader * header, const RegistryEntry & entry
 
   for (uint32_t i = 0; i < max; ++i) {
     uint8_t expected = ENTRY_EMPTY;
-    // CAS state EMPTY -> entry.type. Winner owns the slot.
+    // CAS state EMPTY -> RESERVED. Winner owns the slot, but readers still see
+    // it as empty until we publish the real type below — so no reader can
+    // snapshot the slot before its payload is committed under the seqlock.
     if (slots[i].state.compare_exchange_strong(
-        expected, static_cast<uint8_t>(entry.type),
+        expected, static_cast<uint8_t>(ENTRY_RESERVED),
         std::memory_order_acq_rel, std::memory_order_relaxed))
     {
       write_slot_payload(&slots[i], entry);
+      // Payload is committed (seq even). Publish the real type with release
+      // ordering so any reader that now sees it also sees the payload.
+      slots[i].state.store(static_cast<uint8_t>(entry.type), std::memory_order_release);
       header->generation.fetch_add(1, std::memory_order_acq_rel);
       return static_cast<int32_t>(i);
     }
@@ -347,6 +354,11 @@ void registry_cleanup_stale(RegistryHeader * header)
     if (!snapshot_slot(&slots[i], &snap)) {
       continue;
     }
+    if (snap.pid == 0) {
+      // pid==0 is never a real entry — only a transient snapshot of a slot
+      // whose payload is not yet written. Never reclaim it.
+      continue;
+    }
     if (pid_is_alive_or_unreachable(snap.pid)) {
       continue;
     }

rmw_unix_socket_cpp/src/registry.hpp

@@ -27,6 +27,11 @@ enum RegistryEntryType : uint8_t
   ENTRY_SUBSCRIPTION,
   ENTRY_SERVICE,
   ENTRY_CLIENT,
+  // Transient claim state: a writer won the slot but has not yet committed its
+  // payload. Readers treat it like ENTRY_EMPTY so they never observe a slot
+  // before its payload is published. Never stored in a RegistryEntry; lives
+  // only in the slot's atomic state field, so it costs no struct layout.
+  ENTRY_RESERVED = 0xFF,
 };
 
 // User-facing struct, passed to registry_add. Plain data, no atomics.
@@ -52,11 +57,12 @@ struct RegistryEntry
 // seqlock + atomic state machine so readers never need a global mutex.
 //
 // Protocol (writer):
-//   1. CAS state EMPTY -> target_type    (claims the slot atomically)
+//   1. CAS state EMPTY -> RESERVED       (claims the slot; readers treat as EMPTY)
 //   2. seq.fetch_add(1, acq_rel)         -> seq becomes odd (write in progress)
 //   3. memcpy payload fields
 //   4. seq.fetch_add(1, acq_rel)         -> seq becomes even (snapshot stable)
-//   5. generation.fetch_add(1, acq_rel)  -> wakes cached readers
+//   5. state.store(target_type, release) -> publishes the committed payload
+//   6. generation.fetch_add(1, acq_rel)  -> wakes cached readers
 //
 // Protocol (reader):
 //   1. s1 = seq.load(acquire); if (s1 & 1) skip slot (writer in progress)