Fix records_yaml startup lookup race (#13538)

The records_yaml AuTest intermittently reported that
proxy.config.diags.output.note was missing when a configuration
callback ran while startup was still registering metrics.
Metrics::find() returned the then-current end iterator for a miss, but
RecLookupRecord() compared it with a new end iterator. An intervening
registration made them differ and misread the missing string record as
a numeric metric.

This patch addresses the race by using the direct metrics lookup API,
which reports a miss without comparing two moving end positions. It also
adds a concurrent registration test that exercises string record lookups
during metric creation.

Author: bneradt

src/records/RecCore.cc

@@ -516,18 +516,20 @@ RecGetRecordCounter(const char *name, bool lock)
 RecErrT
 RecLookupRecord(const char *name, void (*callback)(const RecRecord *, void *), void *data, bool lock)
 {
-  RecErrT      err     = REC_ERR_FAIL;
-  ts::Metrics &metrics = ts::Metrics::instance();
-  auto         it      = metrics.find(name);
+  RecErrT             err     = REC_ERR_FAIL;
+  ts::Metrics        &metrics = ts::Metrics::instance();
+  ts::Metrics::IdType metric_id;
 
-  if (it != metrics.end()) {
+  // A metric's storage is stable after creation. Avoid find()/end() here because end() is the current insertion position and
+  // can advance between those two calls while another thread registers a metric.
+  if (auto *metric = metrics.lookup(name, &metric_id); metric != nullptr) {
     RecRecord r{};
-    auto &&[name, type, val] = *it;
 
     r.rec_type     = RECT_PLUGIN;
-    r.data_type    = type == ts::Metrics::MetricType::COUNTER ? RECD_COUNTER : RECD_INT;
-    r.name         = name.data();
-    r.data.rec_int = val;
+    r.data_type    = metrics.type(metric_id) == ts::Metrics::MetricType::COUNTER ? RECD_COUNTER : RECD_INT;
+    r.name         = name;
+    r.data.rec_int = metric->load();
+    r.registered   = true;
 
     callback(&r, data);
     err = REC_ERR_OKAY;

src/records/unit_tests/test_RecRegister.cc

@@ -22,8 +22,12 @@
 #include "iocore/eventsystem/EventSystem.h"
 #include "iocore/eventsystem/RecProcess.h"
 #include "tscore/Layout.h"
+#include "tsutil/Metrics.h"
 #include "test_Diags.h"
 
+#include <atomic>
+#include <thread>
+
 TEST_CASE("RecRegisterConfig - Type Dispatch", "[librecords][RecConfig]")
 {
   SECTION("RecRegisterConfigInt")
@@ -87,3 +91,32 @@ TEST_CASE("RecRegisterStat - Type Dispatch", "[librecords][RecStat]")
     REQUIRE(value == 500);
   }
 }
+
+TEST_CASE("RecLookupRecord - Concurrent metric registration", "[librecords][RecLookup]")
+{
+  constexpr char record_name[]  = "proxy.test.concurrent.string_value";
+  constexpr char record_value[] = "stable";
+
+  REQUIRE(RecRegisterConfigString(RECT_CONFIG, record_name, record_value, RECU_DYNAMIC, RECC_NULL, nullptr, REC_SOURCE_NULL) ==
+          REC_ERR_OKAY);
+
+  std::atomic<bool> finished{false};
+  std::thread       register_metrics([&]() {
+    for (int i = 0; i < 100000; ++i) {
+      ts::Metrics::Counter::createSpan(1);
+    }
+    finished.store(true, std::memory_order_release);
+  });
+
+  bool all_lookups_succeeded = true;
+
+  do {
+    if (RecGetRecordStringAlloc(record_name) != record_value) {
+      all_lookups_succeeded = false;
+      break;
+    }
+  } while (!finished.load(std::memory_order_acquire));
+  register_metrics.join();
+
+  CHECK(all_lookups_succeeded);
+}