diff --git a/docs/storage/change-feed.md b/docs/storage/change-feed.md index 0cb2ac2..83ae8bb 100644 --- a/docs/storage/change-feed.md +++ b/docs/storage/change-feed.md @@ -26,7 +26,7 @@ record CommitEvent(long generation, long txnId, long wallTime, int branch, * `members` is the workspace's `memberChanges()` list in emission order: `MemberChange.Added(edge, incidence, locator)`, `Removed(edge, incidence, locator)`, `Updated(edge, before, after, locatorBefore, locatorAfter)`. The locator is the member id for `SET` edges and the order label (position key in the order tree) for `ORDERED` edges (`Hyperedge.locator`). * `slots` is the workspace's `slotChanges()`: one `SlotChange(slot, key, before, after)` per effective write to a **primary** slot (`Workspace.write` records nothing when `before.equals(after)` and nothing for derived slots). Derived indexes are not in the feed; consumers that need them recompute or read the tree. -`CommitEvent.isEmpty()` is true when both lists are empty. Empty events are never logged: `createBranch`, `closeBranch` and compaction (`rewrite`) commits advance the generation but leave no feed entry. Consumers must therefore treat generation numbers in the feed as **increasing, not contiguous**. +`CommitEvent.isEmpty()` is true when both lists are empty. Empty events are never logged: `createBranch`, `closeBranch` and compaction (`relocate`) commits advance the generation but leave no feed entry. Consumers must therefore treat generation numbers in the feed as **increasing, not contiguous**. ## 2. When events are written diff --git a/docs/storage/maintenance.md b/docs/storage/maintenance.md index 6eaf812..a731a6f 100644 --- a/docs/storage/maintenance.md +++ b/docs/storage/maintenance.md @@ -87,14 +87,14 @@ Liveness is exposed as `StorageEngine.liveness()` (live bytes per segment). The `Compactor.compact(liveThreshold)` (default threshold `compaction_live_ratio = 0.5`): -1. **Choose victims.** A victim is a `SEALED` segment, not the active one, with `live bytes < allocated bytes × threshold`, so the ratio is *live bytes / bytes written*. Candidates are ordered from emptiest to fullest. The background pass takes only the first one (`compact(threshold, 1)`), which bounds how long a single relocation holds the commit lock; `COMPACT` takes all of them. If there are no victims, it reports and returns. +1. **Choose victims.** A victim is a `SEALED` segment, not the active one, with `live bytes < allocated bytes × threshold`, so the ratio is *live bytes / bytes written*. Candidates are ordered from emptiest to fullest. The background pass takes only the first one (`compact(threshold, 1)`); `COMPACT` takes all of them. If there are no victims, it reports and returns. 2. Mark the victims `COMPACTING`. -3. **Relocate.** For every **active** branch of the current generation, call `transactions.rewrite(branch, roots -> roots.map(relocate))`. `TreeWalker.relocate(ref, schema, moving, scope)` rebuilds as a fresh `Ref.Pending` copy: - * every stored node whose page id lies in a victim segment; - * every ancestor of such a node; - * every leaf whose nested trees changed. +3. **Relocate** (`TransactionManager.relocate(moving)`), in three steps so that commits keep going while pages are copied: + 1. **Snapshot**, under the commit lock for a moment: take the latest appended generation and reserve a transaction id. + 2. **Copy**, without the commit lock. For every **active** branch of the snapshot, `TreeWalker.relocate(ref, schema, moving, scope, reached)` rebuilds as a fresh `Ref.Pending` copy every stored node whose page id lies in a victim segment, every ancestor of such a node, and every leaf whose nested trees changed. Subtrees that touch no victim are returned unchanged and stay shared, and every page the walk reaches is recorded. The copies are materialized and written to the active segment right away, with their WAL records kept for later under the reserved transaction id. `TreeWalker.pairMoves` then walks old and new trees side by side to map each old page id to its copy. + 3. **Swap**, under the commit lock, one branch at a time. `TreeWalker.adopt` walks the branch's **latest** roots: a page with a copy is replaced by the copy; a page that existed at the snapshot without a copy is kept as it is, since nothing below it moved; and a page written after the snapshot, by a commit that ran during the copy, is descended into and rebuilt if anything below it moved. Only that last kind costs work under the lock, and there are only as many as commits wrote during the copy. The result is committed like any other commit, carrying the copies' WAL records, and `relocationFence` moves to its generation. Branches created during the copy are swapped the same way. - Subtrees that touch no victim are returned unchanged and stay shared. `rewrite` publishes the new root vector as a normal commit: it takes the commit lock, materializes the pending nodes into the active segment, appends the WAL records and goes through the group-commit barrier. It also sets `relocationFence` to the new generation id. + Without the swap, a crash leaves the copies as unreferenced pages in the active segment, which a later compaction reclaims; the data is unchanged. 4. Mark the victims `RETIRED` with `retiredAt = generation`. Retained history is **not** truncated. Generations older than `retiredAt` keep pointing at the old images in the victims, which stay on disk and readable, so `AT GENERATION` and `AS OF` keep working across a compaction. The victims are reclaimed once `history_limit` has rolled past `retiredAt`. `StorageEngine.compact()` then checkpoints, which persists the new segment states and roots, and calls `reclaim()`. diff --git a/docs/transactions/transactions.md b/docs/transactions/transactions.md index 230e07f..b3cc4cc 100644 --- a/docs/transactions/transactions.md +++ b/docs/transactions/transactions.md @@ -228,7 +228,7 @@ Under `SNAPSHOT` both commits succeed (R replays `Insert(p)` into B, which L nev | Data page writes into the OS page cache | Waking the committing thread | | WAL append of `Commit`, `head.set(next)`, `committer.submit(pending)` | | -The same lock guards `system(...)` (internal writes such as the dictionary), `createBranch`, `closeBranch` and `rewrite` (compaction). `exclusive(action)` takes the lock **and** drains the committer, which is how `Checkpointer.checkpoint()` obtains a quiescent point where `current == head` and every appended commit is durable and published. +The same lock guards `system(...)` (internal writes such as the dictionary), `createBranch`, `closeBranch` and the swap step of `relocate` (compaction), which holds it only to point the latest roots at the already-written copies. `exclusive(action)` takes the lock **and** drains the committer, which is how `Checkpointer.checkpoint()` obtains a quiescent point where `current == head` and every appended commit is durable and published. Because the lock is released before waiting for durability, the next committer can append while the previous batch is being synced. This pipelining is what lets group commit amortise `fsync`. @@ -344,10 +344,10 @@ Had T2 also touched `m1`, or had `E` been an `ORDERED` edge, the overlap check w ## 10. Branch operations and internal writes -`system(branchId, work)`, `createBranch`, `closeBranch` and `rewrite` reuse `append` and the group committer, so they are durable and ordered exactly like user commits: +`system(branchId, work)`, `createBranch`, `closeBranch` and `relocate` reuse `append` and the group committer, so they are durable and ordered exactly like user commits: * `createBranch` appends a `BranchMeta` record and a branch whose roots are the parent's current roots (copy-on-write makes this O(number of slots)). Names must be unique among active branches; the new id is one past the largest existing id. * `closeBranch` (`MERGED` or `DROPPED`, never for `MAIN`) empties the root vector so the branch's pages become unreachable. -* `rewrite` (compaction) replaces roots without member or slot changes and sets the relocation fence. +* `relocate` (compaction) copies pages outside the lock, then replaces roots without member or slot changes and sets the relocation fence ([maintenance.md](../storage/maintenance.md#compaction)). These commits carry no member or slot changes, so their `CommitEvent` is empty and nothing is written to the change feed (see [change feed](../storage/change-feed.md)). diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java index 89f0a6a..1fe2845 100644 --- a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java +++ b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java @@ -1,6 +1,5 @@ package io.hstore.engine.maintenance; -import io.hstore.engine.catalog.Branch; import io.hstore.engine.catalog.Generation; import io.hstore.engine.catalog.SlotRegistry; import io.hstore.engine.page.PageId; @@ -8,7 +7,6 @@ import io.hstore.engine.page.SegmentInfo; import io.hstore.engine.page.SegmentState; import io.hstore.engine.tree.TreeWalker; -import io.hstore.engine.tree.WriteScope; import io.hstore.engine.txn.TransactionManager; import java.util.Comparator; @@ -69,16 +67,9 @@ public Report compact(double liveThreshold, int maxVictims) { victims.forEach(id -> pages.transition(id, SegmentState.COMPACTING, 0)); Set victimSet = Set.copyOf(victims); LongPredicate moving = pageId -> victimSet.contains(PageId.segmentOf(pageId)); - TreeWalker walker = new TreeWalker(transactions.source()); - List branches = transactions.current().branches().values().stream() - .filter(Branch::isActive).map(Branch::id).toList(); - for (int branch : branches) { - generation = transactions.rewrite(branch, roots -> roots.map((slot, ref) -> - walker.relocate(ref, slots.slot(slot).schema(), moving, new WriteScope()))); - } - long retiredAt = generation; + long retiredAt = transactions.relocate(moving); victims.forEach(id -> pages.transition(id, SegmentState.RETIRED, retiredAt)); - return new Report(live, victims, generation); + return new Report(live, victims, retiredAt); } public List reclaim() { diff --git a/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java b/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java index 67233f5..8d3bef7 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java @@ -1,6 +1,9 @@ package io.hstore.engine.tree; +import java.util.ArrayList; +import java.util.List; import java.util.Map; +import java.util.function.LongConsumer; import java.util.function.LongPredicate; public final class TreeWalker { @@ -42,24 +45,100 @@ private void visit(Ref ref, TreeSchema schema, int height, Map } public Ref relocate(Ref root, TreeSchema schema, LongPredicate moving, WriteScope scope) { - return relocate(root, schema, UNKNOWN_HEIGHT, moving, scope); + return relocate(root, schema, moving, scope, _ -> { + }); } - private Ref relocate(Ref ref, TreeSchema schema, int height, LongPredicate inVictim, WriteScope scope) { + public Ref relocate(Ref root, TreeSchema schema, LongPredicate moving, WriteScope scope, LongConsumer reached) { + return relocate(root, schema, UNKNOWN_HEIGHT, moving, scope, reached); + } + + private Ref relocate(Ref ref, TreeSchema schema, int height, LongPredicate inVictim, WriteScope scope, LongConsumer reached) { if (!(ref instanceof Ref.Stored(long pageId, int _, Summary _))) { return ref; } + reached.accept(pageId); boolean moving = inVictim.test(pageId); if (height == 0 && !moving && !schema.codec().holdsRefs()) { return ref; } + return rebuild(ref, pageId, schema, moving, scope, (child, childSchema, childHeight) -> + relocate(child, childSchema, childHeight, inVictim, scope, reached)); + } + + public void pairMoves(Ref before, Ref after, TreeSchema schema, Map moved) { + pairMoves(before, after, schema, UNKNOWN_HEIGHT, moved); + } + + private void pairMoves(Ref before, Ref after, TreeSchema schema, int height, Map moved) { + if (!(before instanceof Ref.Stored(long oldId, int _, Summary _)) || !(after instanceof Ref.Stored(long newId, int _, Summary _)) + || oldId == newId || moved.putIfAbsent(oldId, after) != null) { + return; + } + if (height == 0 && !schema.codec().holdsRefs()) { + return; + } + switch (source.load(oldId, schema)) { + case Branch branch -> { + Branch copy = (Branch) source.load(newId, schema); + for (int i = 0; i < branch.size(); i++) { + pairMoves(branch.child(i), copy.child(i), schema, branch.height() - 1, moved); + } + } + case Leaf leaf -> { + Leaf copy = (Leaf) source.load(newId, schema); + for (int i = 0; i < leaf.size(); i++) { + List oldRefs = new ArrayList<>(); + List newRefs = new ArrayList<>(); + List> schemas = new ArrayList<>(); + mapNested(schema, leaf.value(i), (nested, nestedSchema) -> { + oldRefs.add(nested); + schemas.add(nestedSchema); + return nested; + }); + mapNested(schema, copy.value(i), (nested, _) -> { + newRefs.add(nested); + return nested; + }); + for (int k = 0; k < oldRefs.size(); k++) { + pairMoves(oldRefs.get(k), newRefs.get(k), schemas.get(k), UNKNOWN_HEIGHT, moved); + } + } + } + } + } + + public Ref adopt(Ref root, TreeSchema schema, Map moved, LongPredicate existed, WriteScope scope) { + return adopt(root, schema, UNKNOWN_HEIGHT, moved, existed, scope); + } + + private Ref adopt(Ref ref, TreeSchema schema, int height, Map moved, LongPredicate existed, WriteScope scope) { + if (!(ref instanceof Ref.Stored(long pageId, int _, Summary _))) { + return ref; + } + Ref copy = moved.get(pageId); + if (copy != null) { + return copy; + } + if (existed.test(pageId) || (height == 0 && !schema.codec().holdsRefs())) { + return ref; + } + return rebuild(ref, pageId, schema, false, scope, (child, childSchema, childHeight) -> + adopt(child, childSchema, childHeight, moved, existed, scope)); + } + + @FunctionalInterface + private interface ChildMapper { + Ref map(Ref child, TreeSchema schema, int height); + } + + private Ref rebuild(Ref ref, long pageId, TreeSchema schema, boolean force, WriteScope scope, ChildMapper mapper) { return switch (source.load(pageId, schema)) { case Leaf leaf -> { Object[] values = leaf.valueList().toArray(); - boolean changed = moving; + boolean changed = force; for (int i = 0; i < values.length; i++) { - Object mapped = mapNested(schema, values[i], (nested, nestedSchema) -> - relocate(nested, nestedSchema, UNKNOWN_HEIGHT, inVictim, scope)); + Object mapped = mapNested(schema, values[i], (nested, nestedSchema) -> mapper.map(nested, nestedSchema, UNKNOWN_HEIGHT)); changed |= mapped != values[i]; values[i] = mapped; } @@ -67,11 +146,11 @@ private Ref relocate(Ref ref, TreeSchema schema, int height, LongPredicate in } case Branch branch -> { Ref[] children = branch.childArray(); - boolean changed = moving; + boolean changed = force; for (int i = 0; i < children.length; i++) { - Ref moved = relocate(children[i], schema, branch.height() - 1, inVictim, scope); - changed |= moved != children[i]; - children[i] = moved; + Ref mapped = mapper.map(children[i], schema, branch.height() - 1); + changed |= mapped != children[i]; + children[i] = mapped; } yield changed ? new Ref.Pending(Branch.owned(schema, scope.token(), branch.separatorArray(), children, children.length, branch.height())) diff --git a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java index 443f9bc..19bc1ea 100644 --- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java +++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java @@ -21,6 +21,7 @@ import io.hstore.engine.tree.Ref; import io.hstore.engine.tree.Tree; import io.hstore.engine.tree.TreeSchema; +import io.hstore.engine.tree.TreeWalker; import io.hstore.engine.tree.WriteScope; import io.hstore.engine.wal.WalRecord; import io.hstore.engine.wal.WriteAheadLog; @@ -29,11 +30,14 @@ import java.lang.foreign.ValueLayout; import java.util.ArrayList; import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.OptionalLong; +import java.util.Set; import java.util.TreeSet; import java.util.concurrent.ConcurrentSkipListMap; import java.util.concurrent.atomic.AtomicLong; @@ -41,8 +45,8 @@ import java.util.concurrent.atomic.LongAdder; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Function; +import java.util.function.LongPredicate; import java.util.function.Supplier; -import java.util.function.UnaryOperator; import java.util.stream.Stream; public final class TransactionManager { @@ -305,22 +309,61 @@ private static WalRecord meta(long txn, Branch branch) { return new WalRecord.BranchMeta(txn, branch.id(), branch.name(), branch.parent(), branch.baseGeneration(), branch.createdAt(), branch.state()); } - public long rewrite(int branchId, UnaryOperator rewrite) { - Appended appended; + public long relocate(LongPredicate moving) { + Generation snapshot; + long txnId; commitLock.lock(); try { - Generation latest = head.get(); - Branch branch = latest.branch(branchId); - RootVector roots = rewrite.apply(branch.roots()); - if (roots.sameAs(branch.roots())) { - return latest.id(); - } - appended = append(latest, branch.withRoots(roots), txns.getAndIncrement(), List.of(), List.of(), List.of(), Long.MAX_VALUE); - relocationFence = appended.pending().generation().id(); + snapshot = head.get(); + txnId = txns.getAndIncrement(); } finally { commitLock.unlock(); } - return committer.await(appended.pending()).id(); + TreeWalker walker = new TreeWalker(storage.source()); + Set reached = new HashSet<>(); + Map moved = new HashMap<>(); + List records = new ArrayList<>(); + List writes = new ArrayList<>(); + Materializer materializer = materializer(snapshot.id() + 1, txnId, records, writes); + WriteScope copies = new WriteScope(); + snapshot.branches().values().stream().filter(Branch::isActive).forEach(branch -> branch.roots().roots().forEach((slot, ref) -> { + TreeSchema schema = storage.slots().slot(slot).schema(); + Ref copy = materializer.materialize(walker.relocate(ref, schema, moving, copies, reached::add), schema); + walker.pairMoves(ref, copy, schema, moved); + })); + writes.forEach(write -> storage.pages().write(write.pageId(), write.image())); + pagesWritten.add(materializer.pagesWritten()); + Set swapped = new HashSet<>(); + List unlogged = records; + long generation = snapshot.id(); + while (true) { + Appended appended = null; + commitLock.lock(); + try { + Generation latest = head.get(); + Branch branch = latest.branches().values().stream() + .filter(Branch::isActive).filter(candidate -> !swapped.contains(candidate.id())) + .findFirst().orElse(null); + if (branch == null) { + return generation; + } + swapped.add(branch.id()); + WriteScope merge = new WriteScope(); + RootVector roots = branch.roots().map((slot, ref) -> + walker.adopt(ref, storage.slots().slot(slot).schema(), moved, reached::contains, merge)); + if (!roots.sameAs(branch.roots())) { + appended = append(latest, branch.withRoots(roots), unlogged.isEmpty() ? txns.getAndIncrement() : txnId, + List.of(), List.of(), unlogged, Long.MAX_VALUE); + unlogged = List.of(); + relocationFence = appended.pending().generation().id(); + } + } finally { + commitLock.unlock(); + } + if (appended != null) { + generation = committer.await(appended.pending()).id(); + } + } } public T exclusive(Supplier action) { diff --git a/engine/src/test/java/io/hstore/engine/CompactionTest.java b/engine/src/test/java/io/hstore/engine/CompactionTest.java new file mode 100644 index 0000000..0f00557 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/CompactionTest.java @@ -0,0 +1,263 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine; + +import io.hstore.engine.catalog.Branch; +import io.hstore.engine.maintenance.Compactor; +import io.hstore.engine.page.PageId; +import io.hstore.engine.page.SegmentInfo; +import io.hstore.engine.page.SegmentState; +import io.hstore.engine.topology.EdgeKind; +import io.hstore.engine.topology.Incidence; +import io.hstore.engine.tree.TreeWalker; +import io.hstore.engine.txn.CrashPoint; +import io.hstore.engine.txn.Snapshot; +import io.hstore.engine.txn.TxnOptions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.random.RandomGenerator; +import java.util.random.RandomGeneratorFactory; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class CompactionTest { + + private static final int EDGES = 40; + private static final int MEMBERS = 300; + + @TempDir + Path directory; + + private static EngineOptions options() { + return EngineTest.small().withHistoryLimit(4); + } + + private static Map> load(StorageEngine engine) { + Map> oracle = new HashMap<>(); + for (int e = 0; e < EDGES; e++) { + int index = e; + engine.write(txn -> { + long edge = txn.createEdge(EngineTest.EDGE, EdgeKind.SET); + TreeSet members = new TreeSet<>(); + for (int m = 0; m < MEMBERS; m++) { + long node = txn.createNode(EngineTest.NODE, "n-" + index + "-" + m); + txn.insert(edge, node); + members.add(node); + } + oracle.put(edge, members); + return null; + }); + } + oracle.forEach((edge, members) -> engine.write(txn -> { + List doomed = members.stream().filter(member -> member % 3 != 0).toList(); + doomed.forEach(member -> txn.remove(edge, member)); + doomed.forEach(members::remove); + return null; + })); + return oracle; + } + + private static void churn(StorageEngine engine, int branch, Map> oracle, RandomGenerator random, int commits) { + List edges = new ArrayList<>(oracle.keySet()); + for (int i = 0; i < commits; i++) { + long edge = edges.get(random.nextInt(edges.size())); + TreeSet members = oracle.get(edge); + boolean remove = !members.isEmpty() && random.nextBoolean(); + engine.write(TxnOptions.defaults().onBranch(branch), txn -> { + if (remove) { + long member = members.first(); + txn.remove(edge, member); + members.remove(member); + } else { + long node = txn.createNode(EngineTest.NODE, "churn-" + branch + "-" + random.nextLong()); + txn.insert(edge, node); + members.add(node); + } + return null; + }); + } + } + + private static void verify(StorageEngine engine, int branch, Map> oracle) { + try (Snapshot snapshot = engine.snapshot(branch)) { + oracle.forEach((edge, members) -> assertEquals(List.copyOf(members), + snapshot.requireEdge(edge).stream().map(Incidence::member).toList(), "edge " + edge + " on branch " + branch)); + } + } + + private static Map> copy(Map> oracle) { + return oracle.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, entry -> new TreeSet<>(entry.getValue()))); + } + + @Test + void commitsGoThroughWhileCompactionCopiesPages() throws Exception { + try (StorageEngine engine = StorageEngine.open(directory, options())) { + Map> oracle = load(engine); + CountDownLatch committed = new CountDownLatch(1); + AtomicBoolean waited = new AtomicBoolean(); + AtomicReference failure = new AtomicReference<>(); + Thread compactor = Thread.ofPlatform().start(() -> { + try { + engine.transactions().relocate(pageId -> { + if (waited.compareAndSet(false, true)) { + try { + assertTrue(committed.await(10, TimeUnit.SECONDS), "a commit was blocked while compaction copied pages"); + } catch (InterruptedException e) { + throw new IllegalStateException(e); + } + } + return false; + }); + } catch (Throwable e) { + failure.set(e); + } + }); + while (!waited.get()) { + Thread.onSpinWait(); + } + churn(engine, Branch.MAIN, oracle, RandomGeneratorFactory.of("L64X128MixRandom").create(1), 1); + committed.countDown(); + compactor.join(TimeUnit.SECONDS.toMillis(20)); + assertFalse(compactor.isAlive()); + if (failure.get() != null) { + throw new AssertionError(failure.get()); + } + verify(engine, Branch.MAIN, oracle); + } + } + + @Test + void commitsBetweenSnapshotAndSwapArePointedAtTheCopies() throws Exception { + try (StorageEngine engine = StorageEngine.open(directory, options())) { + Map> oracle = load(engine); + Set victims = engine.stats().segments().stream().filter(segment -> segment.state() == SegmentState.SEALED) + .map(SegmentInfo::id).collect(Collectors.toSet()); + assertTrue(victims.size() > 2, "only " + victims.size() + " sealed segments"); + AtomicBoolean started = new AtomicBoolean(); + AtomicReference failure = new AtomicReference<>(); + RandomGenerator random = RandomGeneratorFactory.of("L64X128MixRandom").create(4); + engine.transactions().relocate(pageId -> { + if (started.compareAndSet(false, true)) { + Thread writer = Thread.ofPlatform().start(() -> { + try { + churn(engine, Branch.MAIN, oracle, random, 50); + } catch (Throwable e) { + failure.set(e); + } + }); + try { + writer.join(); + } catch (InterruptedException e) { + throw new IllegalStateException(e); + } + } + return victims.contains(PageId.segmentOf(pageId)); + }); + if (failure.get() != null) { + throw new AssertionError(failure.get()); + } + Map reachable = new HashMap<>(); + TreeWalker walker = new TreeWalker(engine.transactions().source()); + engine.transactions().current().branches().values().forEach(branch -> branch.roots().roots().forEach((slot, ref) -> + walker.visit(ref, engine.slots().slot(slot).schema(), reachable))); + List stale = reachable.keySet().stream().filter(pageId -> victims.contains(PageId.segmentOf(pageId))).toList(); + assertEquals(List.of(), stale, stale.size() + " pages in the current trees still point into compacted segments"); + verify(engine, Branch.MAIN, oracle); + } + } + + @Test + void commitsOnTwoBranchesDuringCompactionAreKeptAndOldSegmentsGoAway() throws Exception { + Map> main; + Map> side; + int branchId; + List compacted; + try (StorageEngine engine = StorageEngine.open(directory, options())) { + main = load(engine); + Branch branch = engine.createBranch("side", Branch.MAIN); + branchId = branch.id(); + side = copy(main); + AtomicReference report = new AtomicReference<>(); + AtomicReference failure = new AtomicReference<>(); + Thread compactor = Thread.ofPlatform().start(() -> { + try { + report.set(engine.compact()); + } catch (Throwable e) { + failure.set(e); + } + }); + RandomGenerator random = RandomGeneratorFactory.of("L64X128MixRandom").create(2); + int rounds = 0; + int duringCompaction = 0; + while (compactor.isAlive() || rounds < 20) { + churn(engine, Branch.MAIN, main, random, 1); + churn(engine, branchId, side, random, 1); + rounds++; + if (compactor.isAlive()) { + duringCompaction++; + } + } + compactor.join(); + assertTrue(duringCompaction > 0, "no commit landed while compaction ran"); + if (failure.get() != null) { + throw new AssertionError(failure.get()); + } + compacted = report.get().compacted(); + assertFalse(compacted.isEmpty(), "nothing was compacted"); + verify(engine, Branch.MAIN, main); + verify(engine, branchId, side); + churn(engine, Branch.MAIN, main, random, 10); + engine.compact(); + Set remaining = engine.stats().segments().stream().map(SegmentInfo::id).collect(Collectors.toSet()); + compacted.forEach(id -> assertFalse(remaining.contains(id), "segment " + id + " was not reclaimed")); + verify(engine, Branch.MAIN, main); + verify(engine, branchId, side); + } + try (StorageEngine engine = StorageEngine.open(directory, options())) { + verify(engine, Branch.MAIN, main); + verify(engine, branchId, side); + } + } + + @ParameterizedTest + @EnumSource(value = CrashPoint.class, names = {"PAGE", "WAL_APPEND", "DATA_WRITE", "COMMIT_APPEND", "DATA_SYNC", "WAL_SYNC", "CATALOG_PUBLISH"}) + void aCrashWhileSwappingInTheCopiesLosesNothing(CrashPoint point) { + AtomicBoolean armed = new AtomicBoolean(); + EngineOptions options = options().withFaults(reached -> { + if (reached == point && armed.compareAndSet(true, false)) { + throw new CrashPoint.SimulatedCrash(point); + } + }); + Map> oracle; + StorageEngine engine = StorageEngine.open(directory, options); + oracle = load(engine); + armed.set(true); + assertThrows(CrashPoint.SimulatedCrash.class, engine::compact); + engine.halt(); + try (StorageEngine reopened = StorageEngine.open(directory, options())) { + verify(reopened, Branch.MAIN, oracle); + churn(reopened, Branch.MAIN, oracle, RandomGeneratorFactory.of("L64X128MixRandom").create(3), 5); + reopened.compact(); + verify(reopened, Branch.MAIN, oracle); + } + } +}