|
|
@@ -735,51 +735,35 @@ grpc::Status DatabaseGrpcImpl::Find(
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- QueryResult result = store_.find(targetCollection, query);
|
|
|
-
|
|
|
- // v2.0 Stage 4 — shadow-read on Find. Same gates as Exists/Get. We
|
|
|
- // also skip when the result came from the WAL-fallback path (v1.x
|
|
|
- // only; LMDB substrate has no equivalent), because that signals
|
|
|
- // MemoryStore returned data from outside its hot set — comparing
|
|
|
- // would produce spurious divergence. Probe compares total_matched,
|
|
|
- // has_more, document count, and per-doc id+version (capped at first
|
|
|
- // 32 docs to keep shadow cost bounded for large pages).
|
|
|
+ // v2.0 Stage 4 — LMDB-first Find. Same gate as Exists/Get. When the
|
|
|
+ // gate is open we serve the scan from doc_store_; WAL-fallback metrics
|
|
|
+ // are zeroed because the LMDB substrate has no equivalent (no
|
|
|
+ // eviction, no WAL replay on read). On LMDB throw, fall back to the
|
|
|
+ // MemoryStore path and log WARN. Read-your-writes is preserved by
|
|
|
+ // construction: Insert ACK is gated on the dual-write mirror commit,
|
|
|
+ // so any ACK'd write is visible to subsequent LMDB scans.
|
|
|
+ QueryResult result;
|
|
|
+ bool used_lmdb = false;
|
|
|
if (!targetCollection.empty() && targetCollection[0] != '_'
|
|
|
- && !result.usedWalFallback
|
|
|
&& service_.mirrorHealthy() && service_.mirrorDriftCount() == 0) {
|
|
|
if (auto* ds = service_.docStore()) {
|
|
|
try {
|
|
|
auto lmdb_scan = ds->scan(targetCollection, query);
|
|
|
- if (lmdb_scan.total_matched != result.totalCount
|
|
|
- || lmdb_scan.has_more != result.hasMore
|
|
|
- || lmdb_scan.documents.size() != result.documents.size()) {
|
|
|
- spdlog::warn("v2.0 shadow-read Find shape divergence coll={} "
|
|
|
- "mem_total={} lmdb_total={} mem_count={} lmdb_count={} "
|
|
|
- "mem_more={} lmdb_more={}",
|
|
|
- targetCollection,
|
|
|
- result.totalCount, lmdb_scan.total_matched,
|
|
|
- result.documents.size(), lmdb_scan.documents.size(),
|
|
|
- result.hasMore, lmdb_scan.has_more);
|
|
|
- } else {
|
|
|
- const size_t cap = std::min<size_t>(result.documents.size(), 32);
|
|
|
- for (size_t i = 0; i < cap; ++i) {
|
|
|
- const auto& m = result.documents[i];
|
|
|
- const auto& l = lmdb_scan.documents[i];
|
|
|
- if (m.id != l.id || m.version != l.version) {
|
|
|
- spdlog::warn("v2.0 shadow-read Find doc divergence coll={} pos={} "
|
|
|
- "mem_id={} mem_ver={} lmdb_id={} lmdb_ver={}",
|
|
|
- targetCollection, i,
|
|
|
- m.id, m.version, l.id, l.version);
|
|
|
- break; // one log per query is enough signal
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
+ result.documents = std::move(lmdb_scan.documents);
|
|
|
+ result.totalCount = lmdb_scan.total_matched;
|
|
|
+ result.hasMore = lmdb_scan.has_more;
|
|
|
+ // WAL-fallback metrics intentionally zero on the LMDB path.
|
|
|
+ used_lmdb = true;
|
|
|
} catch (const std::exception& e) {
|
|
|
- spdlog::warn("v2.0 shadow-read Find query failed coll={}: {}",
|
|
|
+ spdlog::warn("v2.0 Find LMDB scan failed coll={}: {} "
|
|
|
+ "-- falling back to MemoryStore",
|
|
|
targetCollection, e.what());
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
+ if (!used_lmdb) {
|
|
|
+ result = store_.find(targetCollection, query);
|
|
|
+ }
|
|
|
|
|
|
response->set_total_count(result.totalCount);
|
|
|
response->set_has_more(result.hasMore);
|