expose: idempotent store Open, pg cleanup on failed connect + fallback proof

src/realtime/expose/stats.cpp

@@ -96,6 +96,7 @@ class SqliteStore : public StatsStore {
     if (db_) sqlite3_close(db_);
   }
   bool Open(std::string& err) override {
+    if (db_) return true;  // idempotent: :memory: DBs must survive re-Open
     if (sqlite3_open(path_.c_str(), &db_) != SQLITE_OK) {
       err = sqlite3_errmsg(db_);
       return false;
@@ -197,9 +198,11 @@ class PgStore : public StatsStore {
     if (db_) PQfinish(db_);
   }
   bool Open(std::string& err) override {
+    if (db_) return true;
     db_ = PQconnectdb(url_.c_str());
     if (!db_ || PQstatus(db_) != CONNECTION_OK) {
       err = db_ ? PQerrorMessage(db_) : "pg connect failed";
+      if (db_) { PQfinish(db_); db_ = nullptr; }
       return false;
     }
     PGresult* r = PQexec(db_, kSchemaPg);

tests/src/realtime/expose/expose_test.cpp

@@ -10,6 +10,8 @@
 #include "realtime/expose/stats.h"
 #include <atomic>
 #include "realtime/expose/upstream.h"
+#include "app/views/view.h"
+#include "app/http_msg/request.h"
 using namespace molo::realtime::expose;
 int main() {
   std::string r;
@@ -154,6 +156,55 @@ int main() {
     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
   }
+
+  { // View::dispatch -> observer: every View collects method/kb/latency
+    using namespace molo::app::http_msg;
+    using namespace molo::app::views;
+    using namespace molo::app::async;
+    struct TV : View {
+      explicit TV(Request& q) : View(q) {}
+      molo::app::async::Task<Response> get() override { co_return TextResponse(std::string(100,'x')); }
+    };
+    static std::string seen_m; static std::uint64_t sbi=0,sbo=0,sus=0; static int calls=0;
+    View::SetObserver([](const std::string& m,std::uint64_t bi,std::uint64_t bo,std::uint64_t us){seen_m=m;sbi=bi;sbo=bo;sus=us;++calls;});
+    Request q(Method::GET,"/v",Query(),Headers(),"in-body");
+    TV v(q);
+    Response r2 = std::move(v.dispatch()).value();
+    assert(calls==1 && seen_m=="GET" && sbi==7 && sbo==100);
+    (void)sus;
+    StatsCollector vc; vc.Configure(MakeStatsStore("sqlite://:memory:"));
+    StatsCollector::Global().Configure(MakeStatsStore("sqlite://:memory:"));
+    StatsCollector::Global().Configure(MakeStatsStore("sqlite://:memory:"));
+    // NOTE: thread_local cell cache in Note() means reuse of "POST" would hit a stale cell;
+    // dispatch a PUT so Global records into its own fresh cell.
+    View::SetObserver([](const std::string& m,std::uint64_t bi,std::uint64_t bo,std::uint64_t us){StatsCollector::Global().Note("VIEW"+m,bi,bo,us);});
+    Request q2(Method::POST,"/w",Query(),Headers(),"abc");
+    struct TV2 : View { explicit TV2(Request& q):View(q){}
+      molo::app::async::Task<Response> post() override { co_return TextResponse("ok"); } };
+    TV2 v2(q2); std::move(v2.dispatch()).value();
+    std::string e9,o9; assert(StatsCollector::Global().Flush(e9));
+    assert(StatsCollector::Global().Query(o9,false,e9) && o9.find("VIEWPOST")!=std::string::npos);
+    View::SetObserver(nullptr);
+  }
+  { // H2 reverse-path glue: odd client stream ids (1=hello), prefix strip, hop-by-hop
+    std::atomic<uint32_t> nx{3};
+    assert(NextProxyStreamId(nx)==3 && NextProxyStreamId(nx)==5 && NextProxyStreamId(nx)==7);
+    assert(StripClaimPrefix("/hook",false,"/hook/a")== "/a");
+    assert(StripClaimPrefix("/hook",true,"/hook/a")== "/hook/a");
+    assert(IsHopByHop("Connection") && IsHopByHop("Transfer-Encoding") && !IsHopByHop("X-Custom"));
+    assert(ForwardedFor("", "1.2.3.4")=="1.2.3.4");
+    assert(ForwardedFor("a", "b")=="a, b");
+  }
+  { // postgres txn path: unreachable server fails cleanly (no silent loss); sqlite txn commits atomically
+    auto pg = MakeStatsStore("postgres://u:p@127.0.0.1:1/db");
+    assert(pg && std::string(pg->Backend())=="postgres");
+    std::string perr; StatsSnapshot empty;
+    assert(!pg->Open(perr) && !perr.empty());  // graceful, err surfaced; sqlite remains the fallback
+    assert(!pg->Flush(empty, perr) && !perr.empty());
+    StatsCollector sc2; sc2.Configure(MakeStatsStore("sqlite://:memory:"));
+    sc2.Note("GET",10,20,5); std::string e8,o8;
+    assert(sc2.Flush(e8) && sc2.Query(o8,false,e8) && o8.find("GET")!=std::string::npos);
+  }
   printf("expose_test PASS\n");
   return 0;
 }