Mirror: forward only docs the source actually inserted

Previously Mirror::createDocuments built its destination payload from
the original input ($documents), not from the docs the source actually
persisted. In skipDuplicates mode that lets the destination diverge from
the source whenever the source no-ops a write:

  source    has  id=a, name='Original'
  dest      missing id=a
  caller    skipDuplicates(createDocuments([{id: a, name: 'Duplicate'}]))

The source skips a (already present) but the Mirror still forwards
{id: a, name: 'Duplicate'} to the destination, which inserts it — the
two replicas now hold different values for the same row.

The same code path also throws "All documents must have an sequence if
one is set" inside Database::createDocuments when a partially-skipped
batch reaches the destination, because the inserted doc has a sequence
assigned by the source adapter while the skipped doc does not. Mirror
catches and silently logs that exception, so the destination ends up
missing both rows.

Fix: wrap the user's onNext to capture the docs the source actually
persisted, then forward only those to the destination. Skipped docs
are now correctly never mirrored.

Adds testCreateDocumentsSkipDuplicatesDoesNotDivergeDestination —
seeds a row in the source only (bypassing the mirror), then calls
skipDuplicates(createDocuments(...)) with a different value for that
id plus a new row. Asserts source keeps its original value, destination
ends up with only the new row, and the skipped value never reaches
either side.

Author: premtsd-code

src/Database/Mirror.php

@@ -601,12 +601,25 @@ public function createDocuments(
         ?callable $onNext = null,
         ?callable $onError = null,
     ): int {
+        // Capture the docs the source actually persisted (rather than the input)
+        // so that, in skipDuplicates mode, we don't end up forwarding skipped
+        // duplicates to the destination — that would let the destination diverge
+        // from the source whenever the source no-ops a write.
+        /** @var array<Document> $insertedFromSource */
+        $insertedFromSource = [];
+        $captureOnNext = function (Document $doc) use (&$insertedFromSource, $onNext): void {
+            $insertedFromSource[] = $doc;
+            if ($onNext !== null) {
+                $onNext($doc);
+            }
+        };
+
         $modified = $this->source->skipDuplicates(
             fn () => $this->source->createDocuments(
                 $collection,
                 $documents,
                 $batchSize,
-                $onNext,
+                $captureOnNext,
                 $onError,
             ),
             $this->skipDuplicates
@@ -624,10 +637,15 @@ public function createDocuments(
             return $modified;
         }
 
+        // Nothing for the source to mirror — early-out before touching destination.
+        if (empty($insertedFromSource)) {
+            return $modified;
+        }
+
         try {
             $clones = [];
 
-            foreach ($documents as $document) {
+            foreach ($insertedFromSource as $document) {
                 $clone = clone $document;
 
                 foreach ($this->writeFilters as $filter) {

tests/e2e/Adapter/MirrorTest.php

@@ -313,6 +313,95 @@ public function testDeleteMirroredDocument(): void
         $this->assertTrue($database->getDestination()->getDocument('testDeleteMirroredDocument', $document->getId())->isEmpty());
     }
 
+    /**
+     * Regression: when skipDuplicates causes the source to no-op a write,
+     * the Mirror must NOT forward the input document to the destination.
+     * Otherwise the destination would diverge from the source — receiving
+     * the would-be value for a row the source kept unchanged.
+     */
+    public function testCreateDocumentsSkipDuplicatesDoesNotDivergeDestination(): void
+    {
+        $database = $this->getDatabase();
+        $collection = 'mirrorSkipDup';
+
+        $database->createCollection($collection, attributes: [
+            new Document([
+                '$id' => 'name',
+                'type' => Database::VAR_STRING,
+                'required' => true,
+                'size' => Database::LENGTH_KEY,
+            ]),
+        ], permissions: [
+            Permission::create(Role::any()),
+            Permission::read(Role::any()),
+        ], documentSecurity: false);
+
+        // Seed the SOURCE only (bypass the mirror) with the row we want to
+        // skipDuplicates over later. Destination intentionally does NOT have it.
+        $database->getSource()->createDocument($collection, new Document([
+            '$id' => 'dup',
+            'name' => 'Original',
+            '$permissions' => [
+                Permission::read(Role::any()),
+                Permission::create(Role::any()),
+            ],
+        ]));
+
+        // Sanity check setup
+        $this->assertSame(
+            'Original',
+            $database->getSource()->getDocument($collection, 'dup')->getAttribute('name')
+        );
+        $this->assertTrue(
+            $database->getDestination()->getDocument($collection, 'dup')->isEmpty()
+        );
+
+        // Now do skipDuplicates createDocuments via the Mirror with a different
+        // value for 'dup' plus one new doc 'fresh'. The source must skip 'dup'
+        // (already exists) and insert 'fresh'. The Mirror must forward only
+        // 'fresh' to the destination — NOT 'dup' with the would-be value.
+        $database->skipDuplicates(fn () => $database->createDocuments($collection, [
+            new Document([
+                '$id' => 'dup',
+                'name' => 'WouldBe',
+                '$permissions' => [
+                    Permission::read(Role::any()),
+                    Permission::create(Role::any()),
+                ],
+            ]),
+            new Document([
+                '$id' => 'fresh',
+                'name' => 'Fresh',
+                '$permissions' => [
+                    Permission::read(Role::any()),
+                    Permission::create(Role::any()),
+                ],
+            ]),
+        ]));
+
+        // Source: 'dup' untouched, 'fresh' inserted.
+        $this->assertSame(
+            'Original',
+            $database->getSource()->getDocument($collection, 'dup')->getAttribute('name')
+        );
+        $this->assertSame(
+            'Fresh',
+            $database->getSource()->getDocument($collection, 'fresh')->getAttribute('name')
+        );
+
+        // Destination: 'fresh' was forwarded, 'dup' was NOT (would-be value
+        // is the bug — destination should not have 'dup' at all because the
+        // source skipped it and the destination was never told to insert it).
+        $this->assertTrue(
+            $database->getDestination()->getDocument($collection, 'dup')->isEmpty(),
+            'Mirror must not forward source-skipped docs to the destination'
+        );
+        $this->assertSame(
+            'Fresh',
+            $database->getDestination()->getDocument($collection, 'fresh')->getAttribute('name')
+        );
+    }
+
     protected function deleteColumn(string $collection, string $column): bool
     {
         $sqlTable = "`" . self::$source->getDatabase() . "`.`" . self::$source->getNamespace() . "_" . $collection . "`";