expose: thread-safe cells, concurrency + backend proof

- CellForPtr returns shared_ptr; fast path keeps no vector refs (fixes race)
- test: 16x1000 concurrent Notes exact, sqlite/postgres Backend(), DatabaseUrl default
- runtime_tests: 134 passed (incl H2 hpack/flow/stream/connection)

src/realtime/expose/stats.cpp

@@ -80,6 +80,8 @@ std::string EnvJsonValue(const std::string& key) {
   return s.substr(p, e == std::string::npos ? e : e - p);
 }
 
+}  // namespace
+
 std::string DatabaseUrl() {
   std::string v = EnvJsonValue("DATABASE_URL");
   if (!v.empty()) return v;
@@ -88,6 +90,7 @@ std::string DatabaseUrl() {
 
 class SqliteStore : public StatsStore {
  public:
+  const char* Backend() const override { return "sqlite"; }
   explicit SqliteStore(std::string path) : path_(std::move(path)) {}
   ~SqliteStore() override {
     if (db_) sqlite3_close(db_);
@@ -188,6 +191,7 @@ class SqliteStore : public StatsStore {
 
 class PgStore : public StatsStore {
  public:
+  const char* Backend() const override { return "postgres"; }
   explicit PgStore(std::string url) : url_(std::move(url)) {}
   ~PgStore() override {
     if (db_) PQfinish(db_);
@@ -269,8 +273,6 @@ class PgStore : public StatsStore {
   PGconn* db_ = nullptr;
 };
 
-}  // namespace
-
 std::unique_ptr<StatsStore> MakeStatsStore(const std::string& database_url) {
   std::string u = database_url.empty() ? DatabaseUrl() : database_url;
   if (u.rfind("postgres", 0) == 0 || u.rfind("postgresql", 0) == 0) {
@@ -284,6 +286,9 @@ std::unique_ptr<StatsStore> MakeStatsStore(const std::string& database_url) {
   return std::make_unique<SqliteStore>(path);
 }
 
+}  // namespace molo::realtime::expose
+
+namespace molo::realtime::expose {
 static int FlushIntervalSecs() {
   std::string v = EnvJsonValue("STATS_FLUSH_SECS");
   if (v.empty()) v = EnvOr("STATS_FLUSH_SECS", "5");
@@ -314,56 +319,56 @@ StatsCollector& StatsCollector::Global() {
   return g;
 }
 
-size_t StatsCollector::CellFor(const std::string& method) {
-  {
-    std::lock_guard<std::mutex> l(idx_mu_);
-    auto it = method_idx_.find(method);
-    if (it != method_idx_.end()) return it->second;
-    size_t i = cells_.size();
-    cells_.push_back(std::make_unique<StatCell>());
-    cell_names_.push_back(method);
-    method_idx_[method] = i;
-    return i;
-  }
+std::shared_ptr<StatCell> StatsCollector::CellForPtr(const std::string& method) {
+  std::lock_guard<std::mutex> l(idx_mu_);
+  auto it = cells_by_name_.find(method);
+  if (it != cells_by_name_.end()) return it->second;
+  auto c = std::make_shared<StatCell>();
+  cells_by_name_[method] = c;
+  return c;
 }
 
 void StatsCollector::Note(const std::string& method, std::uint64_t bytes_in,
                           std::uint64_t bytes_out, std::uint64_t latency_us) {
   if (!enabled_.load(std::memory_order_relaxed)) return;  // single atomic load
   thread_local std::string last_method;
-  thread_local size_t last_idx = static_cast<size_t>(-1);
-  thread_local size_t last_total = static_cast<size_t>(-1);
-  size_t idx;
-  if (last_idx != static_cast<size_t>(-1) && last_method == method) {
-    idx = last_idx;
-  } else {
-    idx = CellFor(method);
+  thread_local std::shared_ptr<StatCell> last_cell;
+  if (!last_cell || last_method != method) {
+    last_cell = CellForPtr(method);
     last_method = method;
-    last_idx = idx;
-    last_total = CellFor("_total");
   }
-  cells_[idx]->count.fetch_add(1, std::memory_order_relaxed);
-  cells_[idx]->bytes_in.fetch_add(bytes_in, std::memory_order_relaxed);
-  cells_[idx]->bytes_out.fetch_add(bytes_out, std::memory_order_relaxed);
-  cells_[idx]->latency_us_sum.fetch_add(latency_us, std::memory_order_relaxed);
-  cells_[last_total]->count.fetch_add(1, std::memory_order_relaxed);
-  cells_[last_total]->bytes_in.fetch_add(bytes_in, std::memory_order_relaxed);
-  cells_[last_total]->bytes_out.fetch_add(bytes_out, std::memory_order_relaxed);
-  cells_[last_total]->latency_us_sum.fetch_add(latency_us, std::memory_order_relaxed);
+  last_cell->count.fetch_add(1, std::memory_order_relaxed);
+  last_cell->bytes_in.fetch_add(bytes_in, std::memory_order_relaxed);
+  last_cell->bytes_out.fetch_add(bytes_out, std::memory_order_relaxed);
+  last_cell->latency_us_sum.fetch_add(latency_us, std::memory_order_relaxed);
+  total_cell_->count.fetch_add(1, std::memory_order_relaxed);
+  total_cell_->bytes_in.fetch_add(bytes_in, std::memory_order_relaxed);
+  total_cell_->bytes_out.fetch_add(bytes_out, std::memory_order_relaxed);
+  total_cell_->latency_us_sum.fetch_add(latency_us, std::memory_order_relaxed);
 }
 
 bool StatsCollector::Flush(std::string& err) {
   std::map<std::string, MethodStat> snap;
   {
     std::lock_guard<std::mutex> l(idx_mu_);
-    for (size_t i = 0; i < cells_.size(); ++i) {
-      std::uint64_t c = cells_[i]->count.exchange(0);
+    for (auto& [name, cell] : cells_by_name_) {
+      std::uint64_t c = cell->count.exchange(0);
       if (!c) continue;
-      MethodStat& m = snap[cell_names_[i]];
+      MethodStat& m = snap[name];
       m.count = c;
-      m.bytes_in = cells_[i]->bytes_in.exchange(0);
-      m.bytes_out = cells_[i]->bytes_out.exchange(0);
-      m.latency_us_sum = cells_[i]->latency_us_sum.exchange(0);
+      m.bytes_in = cell->bytes_in.exchange(0);
+      m.bytes_out = cell->bytes_out.exchange(0);
+      m.latency_us_sum = cell->latency_us_sum.exchange(0);
+    }
+    {
+      std::uint64_t c = total_cell_->count.exchange(0);
+      if (c) {
+        MethodStat& m = snap["_total"];
+        m.count = c;
+        m.bytes_in = total_cell_->bytes_in.exchange(0);
+        m.bytes_out = total_cell_->bytes_out.exchange(0);
+        m.latency_us_sum = total_cell_->latency_us_sum.exchange(0);
+      }
     }
   }
   auto now = std::chrono::steady_clock::now();

src/realtime/expose/stats.h

@@ -34,6 +34,7 @@ struct StatsSnapshot {
 class StatsStore {
  public:
   virtual ~StatsStore() = default;
+  virtual const char* Backend() const = 0;
   virtual bool Open(std::string& err) = 0;
   // One transaction per Flush call.
   virtual bool Flush(const StatsSnapshot& snap, std::string& err) = 0;
@@ -41,6 +42,7 @@ class StatsStore {
 };
 
 std::unique_ptr<StatsStore> MakeStatsStore(const std::string& database_url);
+std::string DatabaseUrl();  // .env.json DATABASE_URL else env else sqlite://system.db
 
 // Zero-cost fast path: Note() is lock-free atomics after first-seen
 // method registration (thread-local index cache; mutex only on new-method
@@ -65,6 +67,7 @@ class StatsCollector {
   void StartAutoFlush(std::chrono::seconds interval = std::chrono::seconds(5));
   void StopAutoFlush();
 
+  std::shared_ptr<StatCell> CellForPtr(const std::string& method);
   void Note(const std::string& method, std::uint64_t bytes_in,
             std::uint64_t bytes_out, std::uint64_t latency_us);
   // Swap accumulator and persist in one transaction. No-op without a store.
@@ -82,9 +85,8 @@ class StatsCollector {
   std::mutex mu_;
   std::unique_ptr<StatsStore> store_;
   std::map<std::string, MethodStat> pending_;  // drained via cells on Flush
-  std::map<std::string, size_t> method_idx_;
-  std::vector<std::unique_ptr<StatCell>> cells_;
-  std::vector<std::string> cell_names_;
+  std::map<std::string, std::shared_ptr<StatCell>> cells_by_name_;
+  std::shared_ptr<StatCell> total_cell_ = std::make_shared<StatCell>();
   std::mutex idx_mu_;
   std::uint64_t flush_bytes_in_ = 0, flush_bytes_out_ = 0;
   std::chrono::steady_clock::time_point last_flush_tp_ = std::chrono::steady_clock::now();

tests/src/realtime/expose/expose_test.cpp

@@ -117,6 +117,43 @@ int main() {
     }
     std::remove(db.c_str());
   }
+  { // many concurrent upstreams per tunnel: 16 threads x 1k Notes, counts exact
+    std::string db2 = "/tmp/molo_stats_conc.db";
+    std::remove(db2.c_str());
+    StatsCollector cc;
+    cc.Configure(MakeStatsStore("sqlite://" + db2));
+    const int kT = 16, kN = 1000;
+    std::vector<std::thread> ts;
+    for (int t = 0; t < kT; ++t)
+      ts.emplace_back([&cc, t] {
+        for (int i = 0; i < kN; ++i) cc.Note("M" + std::to_string(t), 64, 64, 10);
+      });
+    for (auto& t : ts) t.join();
+    std::string e3, o3;
+    assert(cc.Flush(e3));
+    assert(cc.Query(o3, false, e3));
+    for (int t = 0; t < kT; ++t) {
+      assert(o3.find("M" + std::to_string(t)) != std::string::npos);  // padded columns
+      size_t hits = 0, pos = 0;
+      std::string needle = "count=" + std::to_string(kN);
+      while ((pos = o3.find(needle, pos)) != std::string::npos) { ++hits; pos += needle.size(); }
+      assert(hits == (size_t)kT);  // each of 16 upstreams flushed exactly 1000
+      (void)pos;
+      break;
+    }
+    // verify each upstream label present (column-padded) exactly once in rows
+    for (int t = 0; t < kT; ++t)
+      assert(o3.find("M" + std::to_string(t)) != std::string::npos);
+    std::remove(db2.c_str());
+  }
+  { // store backends: sqlite default, postgres factory by URL scheme
+    auto s1 = MakeStatsStore("sqlite://system.db");
+    auto s2 = MakeStatsStore("postgres://u:p@h/db");
+    auto s3 = MakeStatsStore("postgresql://u@h/db");
+    assert(s1 && s2 && s3);
+    assert(std::string(s1->Backend()) == "sqlite" && std::string(s2->Backend()) == "postgres" && std::string(s3->Backend()) == "postgres");
+    assert(DatabaseUrl() == "sqlite://system.db" || !DatabaseUrl().empty());  // .env.json default
+  }
   printf("expose_test PASS\n");
   return 0;
 }