Index: oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/Compactor.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/Compactor.java	(date 1438698011000)
+++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/Compactor.java	(date 1438759615000)
@@ -94,6 +94,7 @@
         this.cloneBinaries = compactionStrategy.cloneBinaries();
     }
 
+    // TODO is this actually used or can we remove it?
     protected SegmentNodeBuilder process(NodeState before, NodeState after) {
         SegmentNodeBuilder builder = new SegmentNodeBuilder(
                 writer.writeNode(before), writer);
@@ -101,8 +102,25 @@
         return builder;
     }
 
+    // TODO can we remove this and replace its remaining usages?
     public SegmentNodeState compact(NodeState before, NodeState after) {
         SegmentNodeState compacted = process(before, after).getNodeState();
+        writer.flush();
+        return compacted;
+    }
+
+    /**
+     * Compact the differences between a {@code before} and a {@code after}
+     * on top of an {@code onto} state.
+     * @param before  the before state
+     * @param after   the after state
+     * @param onto    the onto state
+     * @return  the compacted state
+     */
+    public SegmentNodeState compact(NodeState before, NodeState after, NodeState onto) {
+        SegmentNodeBuilder builder = new SegmentNodeBuilder(writer.writeNode(onto), writer);
+        after.compareAgainstBaseState(before, new CompactDiff(builder));
+        SegmentNodeState compacted = builder.getNodeState();
         writer.flush();
         return compacted;
     }
Index: oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/FileStore.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/FileStore.java	(date 1438698011000)
+++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/FileStore.java	(date 1438759615000)
@@ -730,7 +730,7 @@
                     existing);
         }
 
-        SegmentNodeState after = compactor.compact(EMPTY_NODE, before);
+        SegmentNodeState after = compactor.compact(EMPTY_NODE, before, EMPTY_NODE);
 
         Callable<Boolean> setHead = new SetHead(before, after, compactor);
         try {
@@ -744,7 +744,8 @@
                 gcMonitor.info("TarMK compaction detected concurrent commits while compacting. " +
                         "Compacting these commits. Cycle {}", cycles);
                 SegmentNodeState head = getHead();
-                after = compactor.compact(after, head);
+                after = compactor.compact(before, head, after);
+                before = head;
                 setHead = new SetHead(head, after, compactor);
             }
             if (!success) {
@@ -752,7 +753,7 @@
                         "{} cycles.", cycles - 1);
                 if (compactionStrategy.getForceAfterFail()) {
                     gcMonitor.info("TarMK compaction force compacting remaining commits");
-                    if (!forceCompact(after, compactor)) {
+                    if (!forceCompact(before, after, compactor)) {
                         gcMonitor.warn("TarMK compaction failed to force compact remaining commits. " +
                                 "Most likely compaction didn't get exclusive access to the store.");
                     }
@@ -766,11 +767,11 @@
         }
     }
 
-    private boolean forceCompact(final SegmentNodeState before, final Compactor compactor) throws Exception {
+    private boolean forceCompact(final NodeState before, final SegmentNodeState onto, final Compactor compactor) throws Exception {
         return compactionStrategy.compacted(new Callable<Boolean>() {
             @Override
             public Boolean call() throws Exception {
-                return new SetHead(getHead(), compactor.compact(before, getHead()), compactor).call();
+                return new SetHead(getHead(), compactor.compact(before, getHead(), onto), compactor).call();
             }
         });
     }
Index: oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/segment/CompactorTest.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/segment/CompactorTest.java	(date 1438698011000)
+++ oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/segment/CompactorTest.java	(date 1438759615000)
@@ -43,11 +43,11 @@
 
             NodeState initial = store.getRoot();
             SegmentNodeState after = compactor
-                    .compact(initial, store.getRoot());
+                    .compact(initial, store.getRoot(), initial);
             Assert.assertEquals(store.getRoot(), after);
 
             addTestContent(store, 1);
-            after = compactor.compact(initial, store.getRoot());
+            after = compactor.compact(initial, store.getRoot(), initial);
             Assert.assertEquals(store.getRoot(), after);
 
         } finally {
Index: oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/segment/file/FileStoreTest.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/segment/file/FileStoreTest.java	(date 1438698011000)
+++ oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/segment/file/FileStoreTest.java	(date 1438759615000)
@@ -21,6 +21,7 @@
 import static junit.framework.Assert.assertEquals;
 import static junit.framework.Assert.assertFalse;
 import static junit.framework.Assert.assertTrue;
+import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
 import static org.apache.jackrabbit.oak.plugins.segment.SegmentVersion.V_11;
 
 import java.io.ByteArrayInputStream;
@@ -35,7 +36,6 @@
 
 import com.google.common.base.Strings;
 import org.apache.jackrabbit.oak.api.Blob;
-import org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState;
 import org.apache.jackrabbit.oak.plugins.segment.Compactor;
 import org.apache.jackrabbit.oak.plugins.segment.RecordId;
 import org.apache.jackrabbit.oak.plugins.segment.Segment;
@@ -114,7 +114,7 @@
 
         Compactor compactor = new Compactor(writer);
         SegmentNodeState compacted =
-                compactor.compact(EmptyNodeState.EMPTY_NODE, head);
+                compactor.compact(EMPTY_NODE, head, EMPTY_NODE);
         store.close();
 
         // First simulate the case where during compaction a reference to the
@@ -142,7 +142,7 @@
         assertTrue(store.size() > largeBinarySize);
         writer = new SegmentWriter(store, store.getTracker(), V_11);
         compactor = new Compactor(writer);
-        compacted = compactor.compact(EmptyNodeState.EMPTY_NODE, head);
+        compacted = compactor.compact(EMPTY_NODE, head, EMPTY_NODE);
         builder = head.builder();
         builder.setChildNode("old", head); // reference to pre-compacted state
         builder.getNodeState();