diff --git a/src/realtime/expose/stats.cpp b/src/realtime/expose/stats.cpp index 0bd480a..eb82524 100644 --- a/src/realtime/expose/stats.cpp +++ b/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); diff --git a/tests/src/realtime/expose/expose_test.cpp b/tests/src/realtime/expose/expose_test.cpp index 71ce43b..06e0c95 100644 --- a/tests/src/realtime/expose/expose_test.cpp +++ b/tests/src/realtime/expose/expose_test.cpp @@ -10,6 +10,8 @@ #include "realtime/expose/stats.h" #include #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 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 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 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; }