|
@@ -240,9 +240,8 @@ std::string MemoryStore::insert(const std::string& collection, Document doc) {
|
|
|
|
|
|
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
|
|
|
- std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
-
|
|
|
|
|
- // Generate ID if not provided
|
|
|
|
|
|
|
+ // Resolve the ID BEFORE taking the write lock so the in-flight-write
|
|
|
|
|
+ // guard can be marked with the concrete ID that eviction will see.
|
|
|
if (doc.id.empty()) {
|
|
if (doc.id.empty()) {
|
|
|
if (coll->options.autoCreateId) {
|
|
if (coll->options.autoCreateId) {
|
|
|
doc.id = generateId();
|
|
doc.id = generateId();
|
|
@@ -251,6 +250,12 @@ std::string MemoryStore::insert(const std::string& collection, Document doc) {
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ // v1.7.0 T6 quiesce: mark this doc as having an in-flight write so a
|
|
|
|
|
+ // concurrent evictOneChunk() skips it.
|
|
|
|
|
+ InFlightWriteGuard inFlight(*coll, doc.id);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
// Check if ID already exists
|
|
// Check if ID already exists
|
|
|
if (coll->documents.find(doc.id) != coll->documents.end()) {
|
|
if (coll->documents.find(doc.id) != coll->documents.end()) {
|
|
|
throw std::runtime_error("Document with ID '" + doc.id + "' already exists");
|
|
throw std::runtime_error("Document with ID '" + doc.id + "' already exists");
|
|
@@ -372,6 +377,9 @@ bool MemoryStore::update(const std::string& collection, const std::string& id, c
|
|
|
|
|
|
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
|
|
|
|
|
+ // v1.7.0 T6 quiesce: mark before taking the write lock.
|
|
|
|
|
+ InFlightWriteGuard inFlight(*coll, id);
|
|
|
|
|
+
|
|
|
std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
|
|
|
auto it = coll->documents.find(id);
|
|
auto it = coll->documents.find(id);
|
|
@@ -451,9 +459,8 @@ bool MemoryStore::update(const std::string& collection, const std::string& id, c
|
|
|
std::string MemoryStore::upsert(const std::string& collection, Document doc) {
|
|
std::string MemoryStore::upsert(const std::string& collection, Document doc) {
|
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
|
|
|
- std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
-
|
|
|
|
|
- // Generate ID if not provided
|
|
|
|
|
|
|
+ // Resolve the ID BEFORE taking the write lock so the in-flight-write
|
|
|
|
|
+ // guard can be marked with the concrete ID that eviction will see.
|
|
|
if (doc.id.empty()) {
|
|
if (doc.id.empty()) {
|
|
|
if (coll->options.autoCreateId) {
|
|
if (coll->options.autoCreateId) {
|
|
|
doc.id = generateId();
|
|
doc.id = generateId();
|
|
@@ -462,6 +469,12 @@ std::string MemoryStore::upsert(const std::string& collection, Document doc) {
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ // v1.7.0 T6 quiesce: mark before taking the write lock. Closes the
|
|
|
|
|
+ // find-existing-then-write-new-version race with concurrent eviction.
|
|
|
|
|
+ InFlightWriteGuard inFlight(*coll, doc.id);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
// Extract vector before storing document (strips _vector from JSON data)
|
|
// Extract vector before storing document (strips _vector from JSON data)
|
|
|
auto vec = extractVector(*coll, doc.data);
|
|
auto vec = extractVector(*coll, doc.data);
|
|
|
|
|
|
|
@@ -560,6 +573,9 @@ std::string MemoryStore::upsert(const std::string& collection, Document doc) {
|
|
|
bool MemoryStore::remove(const std::string& collection, const std::string& id) {
|
|
bool MemoryStore::remove(const std::string& collection, const std::string& id) {
|
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
|
|
|
|
|
+ // v1.7.0 T6 quiesce: mark before taking the write lock.
|
|
|
|
|
+ InFlightWriteGuard inFlight(*coll, id);
|
|
|
|
|
+
|
|
|
std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
|
|
|
auto it = coll->documents.find(id);
|
|
auto it = coll->documents.find(id);
|
|
@@ -619,6 +635,9 @@ bool MemoryStore::updateIfVersion(const std::string& collection, const std::stri
|
|
|
const Document& doc, uint64_t expectedVersion) {
|
|
const Document& doc, uint64_t expectedVersion) {
|
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
|
|
|
|
|
+ // v1.7.0 T6 quiesce: mark before taking the write lock.
|
|
|
|
|
+ InFlightWriteGuard inFlight(*coll, id);
|
|
|
|
|
+
|
|
|
std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
|
|
|
auto it = coll->documents.find(id);
|
|
auto it = coll->documents.find(id);
|
|
@@ -686,6 +705,9 @@ uint64_t MemoryStore::patchDocument(const std::string& collection, const std::st
|
|
|
const nlohmann::json& patch, const std::string& actor) {
|
|
const nlohmann::json& patch, const std::string& actor) {
|
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
|
|
|
|
|
+ // v1.7.0 T6 quiesce: mark before taking the write lock.
|
|
|
|
|
+ InFlightWriteGuard inFlight(*coll, id);
|
|
|
|
|
+
|
|
|
std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
|
|
|
auto it = coll->documents.find(id);
|
|
auto it = coll->documents.find(id);
|
|
@@ -1965,6 +1987,34 @@ void MemoryStore::clear() {
|
|
|
|
|
|
|
|
// ===== Private Methods =====
|
|
// ===== Private Methods =====
|
|
|
|
|
|
|
|
|
|
+// ----- v1.7.0 T6: per-doc in-flight write guard -----
|
|
|
|
|
+
|
|
|
|
|
+MemoryStore::InFlightWriteGuard::InFlightWriteGuard(CollectionData& coll, std::string docId)
|
|
|
|
|
+ : coll_(&coll), docId_(std::move(docId)) {
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll_->inFlightMutex);
|
|
|
|
|
+ // emplace (default-constructs atomic<uint32_t>(0) on first touch) + increment.
|
|
|
|
|
+ coll_->inFlightWrites[docId_].fetch_add(1, std::memory_order_acq_rel);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+MemoryStore::InFlightWriteGuard::~InFlightWriteGuard() {
|
|
|
|
|
+ if (!coll_) return;
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll_->inFlightMutex);
|
|
|
|
|
+ auto it = coll_->inFlightWrites.find(docId_);
|
|
|
|
|
+ if (it == coll_->inFlightWrites.end()) return;
|
|
|
|
|
+ uint32_t prev = it->second.fetch_sub(1, std::memory_order_acq_rel);
|
|
|
|
|
+ if (prev == 1) {
|
|
|
|
|
+ // Count dropped to 0 — erase to keep the map bounded.
|
|
|
|
|
+ coll_->inFlightWrites.erase(it);
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::hasInFlightWrite(CollectionData& coll, const std::string& docId) {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll.inFlightMutex);
|
|
|
|
|
+ auto it = coll.inFlightWrites.find(docId);
|
|
|
|
|
+ return it != coll.inFlightWrites.end() &&
|
|
|
|
|
+ it->second.load(std::memory_order_acquire) > 0;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
std::string MemoryStore::generateId() const {
|
|
std::string MemoryStore::generateId() const {
|
|
|
static std::random_device rd;
|
|
static std::random_device rd;
|
|
|
static std::mt19937_64 gen(rd());
|
|
static std::mt19937_64 gen(rd());
|
|
@@ -2476,6 +2526,12 @@ uint64_t MemoryStore::evictOneChunk() {
|
|
|
// we don't race with a patch()/update() in flight.
|
|
// we don't race with a patch()/update() in flight.
|
|
|
if (doc.updatedAt > hotWriteFloor) continue;
|
|
if (doc.updatedAt > hotWriteFloor) continue;
|
|
|
|
|
|
|
|
|
|
+ // v1.7.0 T6 quiesce: skip docs with an outstanding write.
|
|
|
|
|
+ // Catches the rare-but-real race where a writer read
|
|
|
|
|
+ // find-existing but hasn't written-new-version yet — a
|
|
|
|
|
+ // 30s hot-write floor can't catch that by timestamp alone.
|
|
|
|
|
+ if (hasInFlightWrite(*collPtr, docId)) continue;
|
|
|
|
|
+
|
|
|
Candidate c;
|
|
Candidate c;
|
|
|
c.collection = collName;
|
|
c.collection = collName;
|
|
|
c.id = docId;
|
|
c.id = docId;
|