feat(tl): pull-based TRANSIENT_LOCAL replay; interest-scoped doorbell

rmw_unix_socket_cpp/src/registry.cpp

@@ -281,7 +281,13 @@ static int32_t try_add_once(RegistryHeader * header, const RegistryEntry & entry
              !header->high_water_slot.compare_exchange_weak(
                cur, want, std::memory_order_relaxed, std::memory_order_relaxed)) {}
       header->generation.fetch_add(1, std::memory_order_acq_rel);
-      ring_doorbells(header);  // strictly after the bump — see ring_doorbells
+      // Strictly after the bump — see ring_doorbells. Doorbell registration
+      // itself rings nobody: the slot is invisible to queries, no consumer
+      // exists for the wake, and ringing here would let K graph-waiting
+      // processes registering at launch produce a K^2/2 datagram mini-storm.
+      if (entry.type != ENTRY_DOORBELL) {
+        ring_doorbells(header);
+      }
       return static_cast<int32_t>(i);
     }
   }
@@ -325,9 +331,16 @@ static void teardown_slot(RegistryEntrySlot * slot)
   slot->seq.fetch_add(1, std::memory_order_acq_rel);
 
   // Unlink outside the seqlock — filesystem op, doesn't need to be observable
-  // by other readers atomically.
+  // by other readers atomically. A TRANSIENT_LOCAL publisher slot names its
+  // latched-cache shm segment here (not a socket file), so a stale-PID reap
+  // reclaims the dead publisher's cache promptly instead of leaking it until
+  // the next rmw_init sweep.
   if (path_copy[0] != '\0') {
-    unlink(path_copy);
+    if (std::strncmp(path_copy, "/ros2_uds_tl_", 13) == 0) {
+      shm_unlink(path_copy);
+    } else {
+      unlink(path_copy);
+    }
   }
 }
 
@@ -351,7 +364,9 @@ void registry_remove(RegistryHeader * header, int32_t index)
   }
   teardown_slot(slot);
   header->generation.fetch_add(1, std::memory_order_acq_rel);
-  ring_doorbells(header);  // strictly after the bump — see ring_doorbells
+  if (st != ENTRY_DOORBELL) {  // doorbell slots have no wake consumers
+    ring_doorbells(header);    // strictly after the bump — see ring_doorbells
+  }
 }
 
 // Best-effort: stat /proc/<pid>. ENOENT means the PID is not in our
@@ -525,7 +540,9 @@ void registry_cleanup_stale(RegistryHeader * header)
 
     teardown_slot(&slots[i]);
     header->generation.fetch_add(1, std::memory_order_acq_rel);
-    reclaimed = true;
+    if (expected != static_cast<uint8_t>(ENTRY_DOORBELL)) {
+      reclaimed = true;  // doorbell-only reclaims ring nobody
+    }
   }
   if (reclaimed) {
     ring_doorbells(header);  // strictly after the bump(s) — see ring_doorbells

rmw_unix_socket_cpp/src/rmw_graph.cpp

@@ -66,8 +66,9 @@ static constexpr int64_t CLEANUP_MIN_INTERVAL_NS = 1000000000;  // 1 s
 // latency: reclaiming a dead process's slots — and with them the doorbell
 // rings and latched-cache teardown only a sweep produces — now lags up to one
 // interval per polling context. The full-registry path in registry_add still
-// sweeps unconditionally, because there it is the last resort before entity
-// creation fails.
+// sweeps unconditionally: there it is the last resort before entity creation
+// fails, and also a recurring retry path (a graph wait whose doorbell
+// registration failed on a full registry retries registry_add every wait).
 static void maybe_cleanup_stale(rmw_uds::UdsContext * ctx, rmw_uds::RegistryHeader * header)
 {
   const int64_t now = steady_now_ns();

rmw_unix_socket_cpp/src/rmw_init.cpp

@@ -204,13 +204,16 @@ rmw_ret_t rmw_init(const rmw_init_options_t * options, rmw_context_t * context)
 
   auto * header = rmw_uds::registry_header(ctx->registry_ptr);
 
-  // Doorbell: bind + register BEFORE the first generation snapshot below, so
-  // no registry mutation can land in the gap between the snapshot and the
-  // wakeup wiring (a mutation after registration rings this socket; one
-  // before it is covered by the snapshot).
+  // Doorbell: bind the socket now so the fd is stable for every rmw_wait, but
+  // do NOT register the ENTRY_DOORBELL slot yet. Registration is lazy, from
+  // the first rmw_wait whose wait set holds a graph guard condition (see
+  // rmw_wait.cpp): with TRANSIENT_LOCAL replay pull-based, only graph-event
+  // consumers need registry wakeups, so plain pub/sub processes never ring —
+  // a fleet launch sends ~zero doorbell datagrams instead of mutations x
+  // processes. The lazy path replicates this bind-before-register ordering.
   {
-    const std::string ctl_path = rmw_uds::make_socket_path(domain_id, "ctl");
-    ctx->doorbell_fd = rmw_uds::create_bound_socket(ctl_path);
+    ctx->doorbell_path = rmw_uds::make_socket_path(domain_id, "ctl");
+    ctx->doorbell_fd = rmw_uds::create_bound_socket(ctx->doorbell_path);
     if (ctx->doorbell_fd < 0) {
       RMW_UDS_LOG_ERROR(
         "rmw_init: failed to create doorbell socket (domain_id=%zu)", domain_id);
@@ -220,23 +223,6 @@ rmw_ret_t rmw_init(const rmw_init_options_t * options, rmw_context_t * context)
       RMW_SET_ERROR_MSG("failed to create doorbell socket");
       return RMW_RET_ERROR;
     }
-    rmw_uds::RegistryEntry dentry;
-    std::memset(&dentry, 0, sizeof(dentry));
-    dentry.type = rmw_uds::ENTRY_DOORBELL;
-    dentry.pid = getpid();
-    std::strncpy(dentry.socket_path, ctl_path.c_str(), sizeof(dentry.socket_path) - 1);
-    ctx->doorbell_registry_index = rmw_uds::registry_add(header, dentry);
-    if (ctx->doorbell_registry_index < 0) {
-      RMW_UDS_LOG_ERROR(
-        "rmw_init: registry full — cannot register doorbell (domain_id=%zu)", domain_id);
-      close(ctx->doorbell_fd);
-      unlink(ctl_path.c_str());
-      close(ctx->send_socket_fd);
-      rmw_uds::registry_close(ctx->registry_fd, ctx->registry_ptr, ctx->registry_size);
-      delete ctx;
-      RMW_SET_ERROR_MSG("registry full — cannot register doorbell");
-      return RMW_RET_ERROR;
-    }
   }
 
   // Read initial generation
@@ -249,7 +235,7 @@ rmw_ret_t rmw_init(const rmw_init_options_t * options, rmw_context_t * context)
   if (options->enclave) {
     enclave_copy = rcutils_strdup(options->enclave, options->allocator);
     if (!enclave_copy) {
-      rmw_uds::registry_remove(header, ctx->doorbell_registry_index);
+      unlink(ctx->doorbell_path.c_str());  // doorbell not yet registered
       close(ctx->doorbell_fd);
       close(ctx->send_socket_fd);
       rmw_uds::registry_close(ctx->registry_fd, ctx->registry_ptr, ctx->registry_size);
@@ -276,7 +262,7 @@ rmw_ret_t rmw_init(const rmw_init_options_t * options, rmw_context_t * context)
     if (enclave_copy) {
       options->allocator.deallocate(enclave_copy, options->allocator.state);
     }
-    rmw_uds::registry_remove(header, ctx->doorbell_registry_index);
+    unlink(ctx->doorbell_path.c_str());  // doorbell not yet registered
     close(ctx->doorbell_fd);
     close(ctx->send_socket_fd);
     rmw_uds::registry_close(ctx->registry_fd, ctx->registry_ptr, ctx->registry_size);
@@ -314,10 +300,13 @@ rmw_ret_t rmw_context_fini(rmw_context_t * context)
   auto * ctx = reinterpret_cast<rmw_uds::UdsContext *>(context->impl);
   if (ctx) {
     // Doorbell teardown before the registry unmaps: registry_remove's slot
-    // teardown also unlinks the socket file.
+    // teardown also unlinks the socket file. When the doorbell was never
+    // lazily registered there is no slot, so unlink the socket file here.
     if (ctx->doorbell_registry_index >= 0 && ctx->registry_ptr) {
       auto * header = rmw_uds::registry_header(ctx->registry_ptr);
       rmw_uds::registry_remove(header, ctx->doorbell_registry_index);
+    } else if (!ctx->doorbell_path.empty()) {
+      unlink(ctx->doorbell_path.c_str());
     }
     if (ctx->doorbell_fd >= 0) {
       close(ctx->doorbell_fd);