diff --git a/CHANGELOG.md b/CHANGELOG.md index 74199d7e0..ad7f0e610 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,27 @@ the live collection had been cleared to zero documents. Both `abortTransaction() whose store was actually built while the transaction was open, not merely the ones it wrote to, since a read-only indexed query can trigger that same lazy rebuild without ever writing. +#### InMemoryDriver: a `CollectionIndexStore` built before a transaction started stayed stale for the whole transaction, silently losing an update on commit +The previous fix only covers a store built DURING a transaction. A store built BEFORE one - +the common case, since most collections already have a store from earlier reads or writes - +was never touched by that invalidation at all. Such a store was built by reading through the +live database and holds live document instances; a transaction's writes then mutate its +private cloned snapshot instead, without that pre-existing store ever finding out. An +index-backed read inside the transaction (an equality lookup on a secondary index) kept +returning the pre-transaction live instance, diverging from a full scan of the same +collection, which does read through the transaction's snapshot. Worse, an update whose +candidate document came from that stale index-backed lookup mutated the live object instead +of the snapshot clone the commit actually merges back, so the write was silently lost after +commit even though it succeeded without error inside the transaction. `getIndexStore()` now +records which transaction context (if any) each persistent store was built from and reuses a +store only for the caller it was built for - rebuilding lazily on first access rather than +eagerly discarding every collection's store at transaction start. Keying this by context +identity rather than by build order matters because `currentTransaction` is thread-local and +transactions genuinely overlap: it stops two concurrent transactions from borrowing each +other's store (which would let one transaction's index-backed update land in the other's +snapshot) and stops a reader outside any transaction from observing an open transaction's +uncommitted writes through a store seeded with that transaction's clones. + #### PoppyDB: a re-syncing secondary broadcast its own initial-sync wipe as change-stream drop events, letting stale watchers destroy `admin.system.users` cluster-wide during a stepdown The initial sync's `clearLocalDatabases()` wipe and snapshot copy ran as regular commands and therefore emitted live change-stream events on the syncing node - including diff --git a/morphium-core/src/main/java/de/caluga/morphium/driver/inmem/InMemoryDriver.java b/morphium-core/src/main/java/de/caluga/morphium/driver/inmem/InMemoryDriver.java index 61eecf579..a18c9f4b5 100644 --- a/morphium-core/src/main/java/de/caluga/morphium/driver/inmem/InMemoryDriver.java +++ b/morphium-core/src/main/java/de/caluga/morphium/driver/inmem/InMemoryDriver.java @@ -344,7 +344,30 @@ private void recordAggregateSlowQueryIfNeeded(String db, String collection, List * gets rebuilt from scratch on the next read - see {@link #getIndexStore} for the lifecycle * contract every write path must follow. */ - private final Map indexStoreByCollection = new ConcurrentHashMap<>(); + private final Map indexStoreByCollection = new ConcurrentHashMap<>(); + + /** + * A {@link CollectionIndexStore} together with the data provenance it was built from: + * either a specific {@link InMemTransactionContext} (the store holds that transaction's + * cloned documents) or {@link #NO_TRANSACTION} (built from the live database). + * + *

Store and owner live in ONE map value on purpose. Held in two parallel maps they could + * not be published atomically, so a concurrent {@link #getIndexStore} on another thread + * could observe a store whose owner entry was not written yet - or already overwritten by a + * third thread - and reuse it for the wrong caller. That is exactly the confusion the owner + * check exists to prevent, so it must not be re-introduced by the bookkeeping itself. + */ + private record OwnedIndexStore(CollectionIndexStore store, Object owner) { + } + + /** + * Sentinel {@link OwnedIndexStore#owner} value marking a store built with no transaction + * active, i.e. one holding live documents. {@code null} is not usable here: it is exactly + * what {@link #currentTransaction}{@code .get()} returns outside a transaction, so a null + * owner could not be told apart from "unknown". + */ + private static final Object NO_TRANSACTION = new Object(); + /** * Counts {@link #buildIndexStore} calls - i.e. full, from-scratch {@code addIndex} rebuilds of @@ -5972,10 +5995,13 @@ private static Object applyElemMatchProjection(String field, Object arrayVal, Ma * Returns the persistent {@link CollectionIndexStore} for {@code db.collection}, building it * on first access from every currently defined non-{@code _id} index * ({@link #isDefaultIdDefinition}) and the collection's current documents - * ({@link CollectionIndexStore#addIndex}). Once built, a store lives forever (until an - * invalidating structural change - see {@link #invalidateIndexStore}) and is kept in sync by - * every write path calling {@code onInsert}/{@code onUpdate}/{@code onRemove} on it directly, - * which is why - unlike Task 3's rebuild-on-miss cache - there is no epoch/version check here. + * ({@link CollectionIndexStore#addIndex}). Once built, a store lives until an invalidating + * structural change (see {@link #invalidateIndexStore}) and is kept in sync by every write + * path calling {@code onInsert}/{@code onUpdate}/{@code onRemove} on it directly, so - unlike + * Task 3's rebuild-on-miss cache - there is no epoch/version check on its CONTENT. There is, + * however, a check on its PROVENANCE: a store is only handed to the caller whose data it was + * built from, since the same map has to serve both live documents and per-transaction clones. + * See the reuse conditions inline below. * *

Lifecycle contract for write paths. A mutation entry point MUST call this method * (or otherwise be sure the store already exists) BEFORE mutating the collection's document @@ -5994,14 +6020,63 @@ private static Object applyElemMatchProjection(String field, Object arrayVal, Ma */ /* package-private */ CollectionIndexStore getIndexStore(String db, String collection) throws MorphiumDriverException { String key = db + "." + collection; - CollectionIndexStore existing = indexStoreByCollection.get(key); + InMemTransactionContext ctx = currentTransaction.get(); + // The provenance this caller requires: its own transaction, or "live" outside one. + Object requiredOwner = ctx == null ? NO_TRANSACTION : ctx; + OwnedIndexStore existing = indexStoreByCollection.get(key); if (existing != null) { - return existing; - } - CollectionIndexStore built = buildIndexStore(db, collection); - CollectionIndexStore prev = indexStoreByCollection.putIfAbsent(key, built); + // A store built before the currently open transaction started is stale: it was + // built by reading through getCollection()/getDB(), which resolves against the LIVE + // database outside a transaction (see buildIndexStore/getDB) - so it holds live + // document instances. startTransaction() then clones the database for this + // transaction's writes to mutate in place, but never told this pre-existing store, + // which keeps serving those now-superseded live instances for the rest of the + // transaction. Index-backed reads inside the transaction see stale data (diverging + // from a full scan, which does resolve against the transaction's snapshot), and any + // update whose candidate came from an index-backed lookup mutates a live object the + // commit never merges back - the write is lost. + // + // Reuse is only safe when the store was built from the same data this caller reads + // through. There are three provenances and, outside a transaction, only one of them + // qualifies: + // + // - NO_TRANSACTION: built from the live database. Valid for a non-transactional + // caller, stale for a transaction (that transaction's writes go to its clones, + // which this store never learns about - the bug this fix exists for). + // - the CALLER's own context: built from exactly the snapshot this caller writes to + // and reads through. Valid for that transaction, and unreachable here for a + // non-transactional caller. + // - SOME OTHER transaction's context: built from a different, possibly still-open + // snapshot holding that transaction's uncommitted clones. Never valid for anyone + // else - a non-transactional reader would observe uncommitted data, and another + // transaction's index-backed update would land in the wrong snapshot, lost on its + // own commit and corrupting the other's on the way. + // + // currentTransaction is thread-local, so transactions genuinely overlap across + // threads (see InMemTransactionIsolationTest) and all three provenances really do + // occur. Build ORDER cannot separate them - a later build may well belong to someone + // else - which is why this is keyed by context identity. + if (existing.owner() == requiredOwner) { + return existing.store(); + } + // Stale (predates this transaction) or foreign (belongs to a different, still-open + // transaction on another thread): fall through to a rebuild, exactly like a cache + // miss. Remove this exact entry (value-compare, so a concurrent replacement by + // another thread is left alone) so a concurrent reader cannot observe it in between. + indexStoreByCollection.remove(key, existing); + } + OwnedIndexStore built = new OwnedIndexStore(buildIndexStore(db, collection), requiredOwner); + // Store and owner are published in a single map operation, so no other thread can ever + // see one without the other. If another thread won the race and published first, its + // entry only counts for us when its provenance matches ours - otherwise we must NOT + // return it (that was the whole point of the check above) and use our own build instead. + // Ours is not published in that case: the winner's entry stays, and the next caller + // re-evaluates provenance normally. Building twice is wasteful but never incorrect (see + // this method's contract), whereas handing back a foreign snapshot's store is exactly + // the cross-transaction leak this guards against. + OwnedIndexStore prev = indexStoreByCollection.putIfAbsent(key, built); if (prev != null) { - return prev; + return prev.owner() == requiredOwner ? prev.store() : built.store(); } // Record that this collection's persistent index store was actually BUILT (not merely // reused) while a transaction is open - see @@ -6010,16 +6085,15 @@ private static Object applyElemMatchProjection(String field, Object arrayVal, Ma // while one is active - i.e. against structurally-cloned document instances, not the // live ones - so only a build can seed the store with clones that must not outlive the // transaction. A plain reuse of an already-built store can never introduce clones: the - // store already existed before this call (built either outside any transaction or by an - // earlier one that has since been invalidated on commit/abort), so it holds only - // references that were valid at the time it was built. Write paths are covered + // identity check above only reuses a store this very transaction built, i.e. one whose + // clones are the ones this transaction is already working on. Write paths are covered // separately and unconditionally by markCollectionTouched before their first store - // mutation, so they need no recording here even though they also call this method. - InMemTransactionContext ctx = currentTransaction.get(); + // mutation, so they need no recording + // here even though they also call this method. if (ctx != null) { ctx.getIndexStoreAccessedCollections().add(db + "/" + collection); } - return built; + return built.store(); } private CollectionIndexStore buildIndexStore(String db, String collection) throws MorphiumDriverException { @@ -10722,6 +10796,13 @@ private void invalidateIndexStoreForKey(String key) { * concurrent non-transactional thread that deletes and then re-inserts a live document under * the same unique key can still collide with the transaction's clone and see a false * duplicate. That race is not introduced by this fix and is not addressed here. + * + *

A separate, single-threaded variant of the general "identity-based staleness" problem + * class - a store built BEFORE the transaction even started, rather than one built during + * it and outliving it - is addressed by {@link #getIndexStore}'s provenance check, not here: + * such a store holds live document instances that this method's touchedCollections/ + * indexStoreAccessedCollections invalidation never sees, because it was never recorded as + * accessed by this (or any) transaction in the first place. */ public void abortTransaction() { InMemTransactionContext ctx = currentTransaction.get(); diff --git a/morphium-core/src/test/java/de/caluga/test/morphium/driver/inmem/InMemTransactionPreExistingIndexStoreStalenessTest.java b/morphium-core/src/test/java/de/caluga/test/morphium/driver/inmem/InMemTransactionPreExistingIndexStoreStalenessTest.java new file mode 100644 index 000000000..90de99b4a --- /dev/null +++ b/morphium-core/src/test/java/de/caluga/test/morphium/driver/inmem/InMemTransactionPreExistingIndexStoreStalenessTest.java @@ -0,0 +1,233 @@ +package de.caluga.test.morphium.driver.inmem; + +import de.caluga.morphium.IndexDescription; +import de.caluga.morphium.driver.Doc; +import de.caluga.morphium.driver.MorphiumDriverException; +import de.caluga.morphium.driver.commands.CreateIndexesCommand; +import de.caluga.morphium.driver.inmem.InMemoryDriver; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * A {@link de.caluga.morphium.driver.inmem.CollectionIndexStore} built before a transaction + * starts is not invalidated by {@code startTransaction()} (unlike a store built DURING one, + * which {@code commitTransaction()}/{@code abortTransaction()} already invalidate - see + * {@code InMemTransactionContext#getIndexStoreAccessedCollections}). Such a pre-existing store + * was built by reading through the live database and therefore holds live document instances, + * while every write inside the transaction mutates the transaction's cloned snapshot instead. + * An index-backed read (equality lookup on a secondary index) inside the transaction then keeps + * returning the pre-transaction live instance - stale relative to a full scan, which does read + * through the transaction's snapshot - and an update whose candidate came from that stale + * index-backed lookup mutates a live object the commit never merges back, so the write is lost. + * + *

Both symptoms are reproduced here: the read-side divergence between an index-backed lookup + * and a full scan while the transaction is still open, and the write loss after commit. + */ +@Tag("inmemory") +public class InMemTransactionPreExistingIndexStoreStalenessTest { + private static final String DB = "testdb"; + private static final String COLL = "uniqcoll"; + + @Test + void preTransactionIndexStore_doesNotSeeUpdateAppliedInsideTransaction() throws Exception { + InMemoryDriver drv = new InMemoryDriver(); + drv.connect(); + try { + new CreateIndexesCommand(drv).setDb(DB).setColl(COLL) + .addIndex(new IndexDescription().setKey(Doc.of("k", 1)).setUnique(true)) + .execute(); + drv.insert(DB, COLL, List.of(Doc.of("_id", 1, "k", "key-1", "status", "created")), + null, true); + + // Force the persistent index store to be built now, strictly BEFORE the + // transaction below starts. This equality lookup on the secondary "k" index is + // exactly the read path CollectionIndexStore.equalityLookup answers. + assertEquals("created", indexLookup(drv).get("status")); + + drv.startTransaction(false); + drv.update(DB, COLL, Doc.of("_id", 1), null, Doc.of("$set", Doc.of("status", "updated")), + false, false, null, null); + + // Read-side symptom: while the transaction is still open, an index-backed lookup + // and a full scan disagree about the very same document. + Map viaIndex = indexLookup(drv); + Map viaFullScan = fullScan(drv); + assertEquals("updated", viaFullScan.get("status"), + "full scan reads through the transaction's snapshot and must see the update"); + assertEquals("updated", viaIndex.get("status"), + "index-backed lookup must agree with the full scan inside the same " + + "transaction instead of still returning the pre-transaction live " + + "document from a store built before the transaction started"); + + drv.commitTransaction(); + + // Write-loss symptom: after commit, the update must be visible however it is read. + assertEquals("updated", fullScan(drv).get("status")); + assertEquals("updated", indexLookup(drv).get("status"), + "the update must survive commit even when read back through the " + + "index-backed path"); + } finally { + drv.shutdown(true); + } + } + + /** + * Two transactions open at the same time on different threads, each with its own cloned + * snapshot ({@code currentTransaction} is thread-local, so this is supported - see + * {@code InMemTransactionIsolationTest}). If the shared store cache were keyed by build + * ORDER rather than by transaction IDENTITY, the transaction that built its store first + * would accept the second transaction's store simply because it was built later. Its + * index-backed update would then mutate the OTHER transaction's clone: lost on its own + * commit, and corrupting the other transaction's snapshot on the way. + */ + @Test + void overlappingTransactions_doNotShareEachOthersIndexStore() throws Exception { + InMemoryDriver drv = new InMemoryDriver(); + drv.connect(); + try { + new CreateIndexesCommand(drv).setDb(DB).setColl(COLL) + .addIndex(new IndexDescription().setKey(Doc.of("k", 1)).setUnique(true)) + .execute(); + drv.insert(DB, COLL, List.of(Doc.of("_id", 1, "k", "key-1", "status", "created")), + null, true); + + // Transaction A on this thread: opens, then builds its store from its own snapshot + // via an index-backed read. + drv.startTransaction(false); + assertEquals("created", indexLookup(drv).get("status")); + + // Transaction B on another thread: opens LATER and builds a store from ITS snapshot, + // then STAYS OPEN. B's store therefore sits in the shared cache, built after A's and + // holding B's clones, at the moment A reaches for it below. B must not commit or + // abort here: either would invalidate the store (see commitTransaction/ + // abortTransaction) and A would simply rebuild, hiding the very confusion under test. + java.util.concurrent.CountDownLatch bBuiltItsStore = new java.util.concurrent.CountDownLatch(1); + java.util.concurrent.CountDownLatch aIsDone = new java.util.concurrent.CountDownLatch(1); + Throwable[] failure = new Throwable[1]; + Thread other = new Thread(() -> { + try { + drv.startTransaction(false); + indexLookup(drv); + drv.update(DB, COLL, Doc.of("_id", 1), null, + Doc.of("$set", Doc.of("status", "from-b")), false, false, null, null); + bBuiltItsStore.countDown(); + aIsDone.await(); + drv.abortTransaction(); + } catch (Throwable t) { + failure[0] = t; + bBuiltItsStore.countDown(); + } + }); + other.start(); + bBuiltItsStore.await(); + if (failure[0] != null) { + throw new AssertionError("transaction B failed", failure[0]); + } + + // Back in A, while B is still open: an index-backed read must NOT see B's write, and + // an index-backed update must land in A's OWN snapshot. If A reused B's store, the + // candidate would be B's clone - A would read "from-b" here and its write would go + // astray. + assertEquals("created", indexLookup(drv).get("status"), + "transaction A must not see an uncommitted write from a concurrently open " + + "transaction through a shared index store"); + drv.update(DB, COLL, Doc.of("_id", 1), null, + Doc.of("$set", Doc.of("status", "from-a")), false, false, null, null); + assertEquals("from-a", indexLookup(drv).get("status"), + "transaction A must read back its own write, not another transaction's"); + assertEquals("from-a", fullScan(drv).get("status")); + + drv.commitTransaction(); + aIsDone.countDown(); + other.join(); + if (failure[0] != null) { + throw new AssertionError("transaction B failed", failure[0]); + } + + assertEquals("from-a", fullScan(drv).get("status"), + "A committed and B aborted, so A's write is the one that must survive"); + assertEquals("from-a", indexLookup(drv).get("status")); + } finally { + drv.shutdown(true); + } + } + + /** + * A reader outside any transaction must never see a still-open transaction's uncommitted + * write, not even when that transaction built the shared index store first and the reader's + * lookup is index-backed. The store built from the transaction's clones is valid only for + * that transaction; anyone else has to get a store built from the live database. + */ + @Test + void nonTransactionalReader_doesNotSeeAnOpenTransactionsUncommittedWrite() throws Exception { + InMemoryDriver drv = new InMemoryDriver(); + drv.connect(); + try { + new CreateIndexesCommand(drv).setDb(DB).setColl(COLL) + .addIndex(new IndexDescription().setKey(Doc.of("k", 1)).setUnique(true)) + .execute(); + drv.insert(DB, COLL, List.of(Doc.of("_id", 1, "k", "key-1", "status", "created")), + null, true); + + // The transaction runs on another thread and stays open, so its store - seeded with + // its own clones - is the one sitting in the shared cache while we read below. + java.util.concurrent.CountDownLatch txHasWritten = new java.util.concurrent.CountDownLatch(1); + java.util.concurrent.CountDownLatch readerIsDone = new java.util.concurrent.CountDownLatch(1); + Throwable[] failure = new Throwable[1]; + Thread tx = new Thread(() -> { + try { + drv.startTransaction(false); + indexLookup(drv); + drv.update(DB, COLL, Doc.of("_id", 1), null, + Doc.of("$set", Doc.of("status", "uncommitted")), false, false, null, null); + txHasWritten.countDown(); + readerIsDone.await(); + drv.abortTransaction(); + } catch (Throwable t) { + failure[0] = t; + txHasWritten.countDown(); + } + }); + tx.start(); + txHasWritten.await(); + if (failure[0] != null) { + throw new AssertionError("the transaction thread failed", failure[0]); + } + + // This thread has no transaction: both read paths must still show the live document. + assertEquals("created", indexLookup(drv).get("status"), + "an index-backed read outside any transaction must not observe an open " + + "transaction's uncommitted write"); + assertEquals("created", fullScan(drv).get("status")); + + readerIsDone.countDown(); + tx.join(); + if (failure[0] != null) { + throw new AssertionError("the transaction thread failed", failure[0]); + } + + // The transaction aborted, so the live document is unchanged. + assertEquals("created", indexLookup(drv).get("status")); + assertEquals("created", fullScan(drv).get("status")); + } finally { + drv.shutdown(true); + } + } + + private Map indexLookup(InMemoryDriver drv) throws MorphiumDriverException { + List> result = drv.find(DB, COLL, Doc.of("k", "key-1"), null, null, 0, 0); + assertEquals(1, result.size()); + return result.get(0); + } + + private Map fullScan(InMemoryDriver drv) throws MorphiumDriverException { + List> result = drv.find(DB, COLL, Doc.of(), null, null, 0, 0); + assertEquals(1, result.size()); + return result.get(0); + } +}