|
|
@@ -0,0 +1,262 @@
|
|
|
+// v2.3 multi-project stress test.
|
|
|
+//
|
|
|
+// Three projects (alpha, beta, gamma) run their own writer + reader
|
|
|
+// fleets concurrently against ONE database instance. Verifies:
|
|
|
+// 1) Throughput holds up (per-project writers ≥ 200/s combined ≥ 600/s)
|
|
|
+// 2) No cross-project leakage — a doc inserted into "alpha" never
|
|
|
+// appears in a "beta" read, even when both target the same doc-id.
|
|
|
+// 3) Zero hard failures (RESOURCE_EXHAUSTED / DEADLINE_EXCEEDED / Other)
|
|
|
+//
|
|
|
+// Each project gets its own Client::Config so the qualify() prefix
|
|
|
+// path is exercised; per-call qualified form is exercised by the
|
|
|
+// integrity checker.
|
|
|
+
|
|
|
+#include <smartbotic/database/client.hpp>
|
|
|
+
|
|
|
+#include <atomic>
|
|
|
+#include <chrono>
|
|
|
+#include <iostream>
|
|
|
+#include <mutex>
|
|
|
+#include <random>
|
|
|
+#include <string>
|
|
|
+#include <thread>
|
|
|
+#include <vector>
|
|
|
+
|
|
|
+using namespace smartbotic::database;
|
|
|
+using Clock = std::chrono::steady_clock;
|
|
|
+
|
|
|
+namespace {
|
|
|
+
|
|
|
+struct Counters {
|
|
|
+ std::atomic<uint64_t> inserts{0};
|
|
|
+ std::atomic<uint64_t> insertFails{0};
|
|
|
+ std::atomic<uint64_t> reads{0};
|
|
|
+ std::atomic<uint64_t> readFound{0};
|
|
|
+ std::atomic<uint64_t> readNotFound{0};
|
|
|
+ std::atomic<uint64_t> readFails{0};
|
|
|
+ std::atomic<uint64_t> crossProjectLeaks{0}; // a doc from another project
|
|
|
+ std::atomic<bool> stop{false};
|
|
|
+};
|
|
|
+
|
|
|
+struct ProjectShard {
|
|
|
+ std::string name;
|
|
|
+ Counters counters;
|
|
|
+ std::mutex idsMu;
|
|
|
+ std::vector<std::string> recentIds;
|
|
|
+ std::vector<std::string> peerIds; // doc-ids that belong to OTHER projects;
|
|
|
+ // used to verify isolation
|
|
|
+};
|
|
|
+
|
|
|
+void pushId(ProjectShard& s, const std::string& id) {
|
|
|
+ std::lock_guard<std::mutex> g(s.idsMu);
|
|
|
+ if (s.recentIds.size() >= 500) s.recentIds.erase(s.recentIds.begin());
|
|
|
+ s.recentIds.push_back(id);
|
|
|
+}
|
|
|
+
|
|
|
+std::optional<std::string> sampleId(ProjectShard& s, std::mt19937& rng) {
|
|
|
+ std::lock_guard<std::mutex> g(s.idsMu);
|
|
|
+ if (s.recentIds.empty()) return std::nullopt;
|
|
|
+ std::uniform_int_distribution<size_t> d(0, s.recentIds.size() - 1);
|
|
|
+ return s.recentIds[d(rng)];
|
|
|
+}
|
|
|
+
|
|
|
+void pushPeerId(ProjectShard& s, const std::string& id) {
|
|
|
+ std::lock_guard<std::mutex> g(s.idsMu);
|
|
|
+ if (s.peerIds.size() >= 500) s.peerIds.erase(s.peerIds.begin());
|
|
|
+ s.peerIds.push_back(id);
|
|
|
+}
|
|
|
+
|
|
|
+std::optional<std::string> samplePeerId(ProjectShard& s, std::mt19937& rng) {
|
|
|
+ std::lock_guard<std::mutex> g(s.idsMu);
|
|
|
+ if (s.peerIds.empty()) return std::nullopt;
|
|
|
+ std::uniform_int_distribution<size_t> d(0, s.peerIds.size() - 1);
|
|
|
+ return s.peerIds[d(rng)];
|
|
|
+}
|
|
|
+
|
|
|
+void writerLoop(ProjectShard& shard, int workerId) {
|
|
|
+ Client::Config cfg;
|
|
|
+ cfg.address = "localhost:9005";
|
|
|
+ cfg.timeoutMs = 5000;
|
|
|
+ cfg.project = shard.name;
|
|
|
+ Client c(cfg);
|
|
|
+ if (!c.connect()) {
|
|
|
+ std::cerr << "[" << shard.name << " writer " << workerId << "] connect failed\n";
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ uint64_t seq = 0;
|
|
|
+ while (!shard.counters.stop.load(std::memory_order_relaxed)) {
|
|
|
+ std::string id = shard.name + "-w" + std::to_string(workerId)
|
|
|
+ + "-" + std::to_string(seq);
|
|
|
+ try {
|
|
|
+ nlohmann::json doc{{"project", shard.name},
|
|
|
+ {"worker", workerId},
|
|
|
+ {"seq", seq}};
|
|
|
+ c.insert("docs", doc, id); // qualifies to "<project>:docs"
|
|
|
+ shard.counters.inserts.fetch_add(1, std::memory_order_relaxed);
|
|
|
+ pushId(shard, id);
|
|
|
+ } catch (const std::exception&) {
|
|
|
+ shard.counters.insertFails.fetch_add(1, std::memory_order_relaxed);
|
|
|
+ }
|
|
|
+ ++seq;
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+void readerLoop(ProjectShard& shard, int workerId) {
|
|
|
+ Client::Config cfg;
|
|
|
+ cfg.address = "localhost:9005";
|
|
|
+ cfg.timeoutMs = 5000;
|
|
|
+ cfg.project = shard.name;
|
|
|
+ Client c(cfg);
|
|
|
+ if (!c.connect()) {
|
|
|
+ std::cerr << "[" << shard.name << " reader " << workerId << "] connect failed\n";
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ std::mt19937 rng(std::random_device{}() + workerId);
|
|
|
+ while (!shard.counters.stop.load(std::memory_order_relaxed)) {
|
|
|
+ // 80% of the time read our own doc; 20% try a peer's id and
|
|
|
+ // verify we get NOT_FOUND (cross-project isolation check).
|
|
|
+ std::uniform_int_distribution<int> dice(0, 9);
|
|
|
+ if (dice(rng) < 8) {
|
|
|
+ auto idOpt = sampleId(shard, rng);
|
|
|
+ if (!idOpt) { std::this_thread::sleep_for(std::chrono::milliseconds(1)); continue; }
|
|
|
+ try {
|
|
|
+ auto doc = c.get("docs", *idOpt);
|
|
|
+ shard.counters.reads.fetch_add(1, std::memory_order_relaxed);
|
|
|
+ if (doc) shard.counters.readFound.fetch_add(1);
|
|
|
+ else shard.counters.readNotFound.fetch_add(1);
|
|
|
+ } catch (const std::exception&) {
|
|
|
+ shard.counters.readFails.fetch_add(1);
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ auto peerOpt = samplePeerId(shard, rng);
|
|
|
+ if (!peerOpt) { std::this_thread::sleep_for(std::chrono::milliseconds(1)); continue; }
|
|
|
+ try {
|
|
|
+ auto doc = c.get("docs", *peerOpt);
|
|
|
+ shard.counters.reads.fetch_add(1);
|
|
|
+ if (doc) {
|
|
|
+ // ISOLATION BREACH — we should NOT see another project's doc.
|
|
|
+ shard.counters.crossProjectLeaks.fetch_add(1);
|
|
|
+ std::cerr << "[" << shard.name << " reader " << workerId
|
|
|
+ << "] LEAK: saw foreign id " << *peerOpt << "\n";
|
|
|
+ } else {
|
|
|
+ shard.counters.readNotFound.fetch_add(1);
|
|
|
+ }
|
|
|
+ } catch (const std::exception&) {
|
|
|
+ shard.counters.readFails.fetch_add(1);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+} // namespace
|
|
|
+
|
|
|
+int main(int argc, char* argv[]) {
|
|
|
+ int durationSec = (argc > 1) ? std::stoi(argv[1]) : 20;
|
|
|
+ int writersPerProject = (argc > 2) ? std::stoi(argv[2]) : 2;
|
|
|
+ int readersPerProject = (argc > 3) ? std::stoi(argv[3]) : 3;
|
|
|
+
|
|
|
+ std::cout << "=== v2.3 multi-project stress ===\n"
|
|
|
+ << "Duration: " << durationSec << "s\n"
|
|
|
+ << "Projects: alpha, beta, gamma\n"
|
|
|
+ << "Writers/proj: " << writersPerProject << "\n"
|
|
|
+ << "Readers/proj: " << readersPerProject << "\n\n";
|
|
|
+
|
|
|
+ // One bootstrap client to ensure projects exist; not strictly required
|
|
|
+ // (per-write would lazy-create) but the explicit createProject also
|
|
|
+ // exercises that RPC.
|
|
|
+ {
|
|
|
+ Client::Config c;
|
|
|
+ c.address = "localhost:9005";
|
|
|
+ Client bootstrap(c);
|
|
|
+ if (!bootstrap.connect()) {
|
|
|
+ std::cerr << "bootstrap connect failed\n";
|
|
|
+ return 1;
|
|
|
+ }
|
|
|
+ for (const char* p : {"alpha", "beta", "gamma"}) {
|
|
|
+ try { bootstrap.createProject(p); }
|
|
|
+ catch (const std::exception& e) {
|
|
|
+ std::cerr << "createProject(" << p << ") failed: " << e.what() << "\n";
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ std::vector<std::unique_ptr<ProjectShard>> shards;
|
|
|
+ for (const auto& name : {"alpha", "beta", "gamma"}) {
|
|
|
+ auto s = std::make_unique<ProjectShard>();
|
|
|
+ s->name = name;
|
|
|
+ shards.push_back(std::move(s));
|
|
|
+ }
|
|
|
+
|
|
|
+ // Cross-pollinate peer ids: each shard's peerIds = all OTHER shards'
|
|
|
+ // ids as they come in. Since shards are concurrent, use a periodic
|
|
|
+ // cross-fill thread.
|
|
|
+ std::vector<std::thread> threads;
|
|
|
+ for (auto& sp : shards) {
|
|
|
+ ProjectShard& s = *sp;
|
|
|
+ for (int w = 0; w < writersPerProject; ++w) threads.emplace_back(writerLoop, std::ref(s), w);
|
|
|
+ for (int r = 0; r < readersPerProject; ++r) threads.emplace_back(readerLoop, std::ref(s), r);
|
|
|
+ }
|
|
|
+
|
|
|
+ // Cross-fill: every 500ms, copy 5 ids from each shard into the OTHER
|
|
|
+ // shards' peerIds list. Readers sample from peerIds for isolation
|
|
|
+ // probes.
|
|
|
+ std::thread crossfill([&]() {
|
|
|
+ while (!shards[0]->counters.stop.load()) {
|
|
|
+ std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
|
|
+ for (size_t i = 0; i < shards.size(); ++i) {
|
|
|
+ std::vector<std::string> snap;
|
|
|
+ {
|
|
|
+ std::lock_guard<std::mutex> g(shards[i]->idsMu);
|
|
|
+ int take = std::min<int>(5, shards[i]->recentIds.size());
|
|
|
+ for (int k = 0; k < take; ++k) snap.push_back(shards[i]->recentIds[k]);
|
|
|
+ }
|
|
|
+ for (size_t j = 0; j < shards.size(); ++j) {
|
|
|
+ if (i == j) continue;
|
|
|
+ for (const auto& id : snap) pushPeerId(*shards[j], id);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ });
|
|
|
+
|
|
|
+ std::this_thread::sleep_for(std::chrono::seconds(durationSec));
|
|
|
+ for (auto& sp : shards) sp->counters.stop.store(true);
|
|
|
+ for (auto& t : threads) t.join();
|
|
|
+ crossfill.join();
|
|
|
+
|
|
|
+ std::cout << "\n=== Per-project results ===\n";
|
|
|
+ bool any_leak = false;
|
|
|
+ bool any_hard_fail = false;
|
|
|
+ uint64_t total_inserts = 0;
|
|
|
+ uint64_t total_reads = 0;
|
|
|
+ for (auto& sp : shards) {
|
|
|
+ ProjectShard& s = *sp;
|
|
|
+ std::cout << "[" << s.name << "]"
|
|
|
+ << " inserts=" << s.counters.inserts.load()
|
|
|
+ << " insertFails=" << s.counters.insertFails.load()
|
|
|
+ << " reads=" << s.counters.reads.load()
|
|
|
+ << " found=" << s.counters.readFound.load()
|
|
|
+ << " not-found=" << s.counters.readNotFound.load()
|
|
|
+ << " readFails=" << s.counters.readFails.load()
|
|
|
+ << " leaks=" << s.counters.crossProjectLeaks.load()
|
|
|
+ << "\n";
|
|
|
+ total_inserts += s.counters.inserts.load();
|
|
|
+ total_reads += s.counters.reads.load();
|
|
|
+ if (s.counters.crossProjectLeaks.load() > 0) any_leak = true;
|
|
|
+ if (s.counters.readFails.load() > 0) any_hard_fail = true;
|
|
|
+ }
|
|
|
+
|
|
|
+ const double total_w_rate = static_cast<double>(total_inserts) / durationSec;
|
|
|
+ const double total_r_rate = static_cast<double>(total_reads) / durationSec;
|
|
|
+
|
|
|
+ std::cout << "\nTotal writers ops/s: " << static_cast<uint64_t>(total_w_rate) << "\n";
|
|
|
+ std::cout << "Total readers ops/s: " << static_cast<uint64_t>(total_r_rate) << "\n";
|
|
|
+
|
|
|
+ std::cout << "\n=== Verdict ===\n";
|
|
|
+ std::cout << " no cross-project leaks: " << (any_leak ? "FAIL" : "PASS") << "\n";
|
|
|
+ std::cout << " no hard read failures: " << (any_hard_fail ? "FAIL" : "PASS") << "\n";
|
|
|
+ std::cout << " total writers > 600/s: "
|
|
|
+ << (total_w_rate > 600 ? "PASS" : "FAIL")
|
|
|
+ << " (actual " << static_cast<uint64_t>(total_w_rate) << ")\n";
|
|
|
+
|
|
|
+ return (any_leak || any_hard_fail || total_w_rate <= 600) ? 1 : 0;
|
|
|
+}
|