Address Jake's follow-up review: tighten limits, reuse tenantKey, let destination handle duplicates

Author: premtsd-code

src/Database/Database.php

@@ -7163,11 +7163,11 @@ public function upsertDocumentsWithIncrease(
                     $found = $this->authorization->skip(fn () => $this->withTenant($tenant, fn () => $this->silent(
                         fn () => $this->find($collection->getId(), [
                             Query::equal('$id', $chunk),
-                            Query::limit(PHP_INT_MAX),
+                            Query::limit($this->maxQueryValues),
                         ])
                     )));
                     foreach ($found as $doc) {
-                        $existingDocs[$tenant . ':' . $doc->getId()] = $doc;
+                        $existingDocs[$this->tenantKey($doc)] = $doc;
                     }
                 }
             }
@@ -7182,7 +7182,7 @@ public function upsertDocumentsWithIncrease(
                     $existing = $this->authorization->skip(fn () => $this->silent(
                         fn () => $this->find($collection->getId(), [
                             Query::equal('$id', $chunk),
-                            Query::limit(PHP_INT_MAX),
+                            Query::limit($this->maxQueryValues),
                         ])
                     ));
                     foreach ($existing as $doc) {

src/Database/Mirror.php

@@ -601,31 +601,6 @@ public function createDocuments(
         ?callable $onNext = null,
         ?callable $onError = null,
     ): int {
-        // In skipDuplicates mode, identify which input ids already exist on source.
-        // These will be silently no-oped by the adapter's INSERT IGNORE and must
-        // not propagate to destination — a skipped duplicate is not a user write.
-        $existingIds = [];
-        if ($this->skipDuplicates) {
-            $ids = \array_values(\array_filter(
-                \array_map(fn (Document $d) => $d->getId(), $documents),
-                fn ($id) => $id !== ''
-            ));
-
-            if (!empty($ids)) {
-                foreach (\array_chunk(\array_unique($ids), \max(1, $this->maxQueryValues)) as $chunk) {
-                    $existing = $this->source->silent(
-                        fn () => $this->source->find($collection, [
-                            Query::equal('$id', $chunk),
-                            Query::limit(PHP_INT_MAX),
-                        ])
-                    );
-                    foreach ($existing as $doc) {
-                        $existingIds[$doc->getId()] = true;
-                    }
-                }
-            }
-        }
-
         $modified = $this->skipDuplicates
             ? $this->source->skipDuplicates(
                 fn () => $this->source->createDocuments($collection, $documents, $batchSize, $onNext, $onError)
@@ -644,22 +619,13 @@ public function createDocuments(
             return $modified;
         }
 
-        // In skipDuplicates mode, drop pre-existing ids so their no-op writes
-        // don't propagate. Non-skip mode forwards everything as before.
-        $toForward = $this->skipDuplicates
-            ? \array_values(\array_filter(
-                $documents,
-                fn (Document $d) => $d->getId() === '' || !isset($existingIds[$d->getId()])
-            ))
-            : $documents;
-
-        if (empty($toForward)) {
-            return $modified;
-        }
-
+        // Forward every input to destination. "upgraded" status means the schema
+        // is mirrored, not that every row is backfilled, so a row that is a
+        // duplicate on source may not yet exist on destination. In skipDuplicates
+        // mode the destination runs its own INSERT IGNORE and decides per-row.
         try {
             $clones = [];
-            foreach ($toForward as $document) {
+            foreach ($documents as $document) {
                 $clone = clone $document;
                 foreach ($this->writeFilters as $filter) {
                     $clone = $filter->beforeCreateDocument(

tests/e2e/Adapter/MirrorTest.php

@@ -313,7 +313,7 @@ public function testDeleteMirroredDocument(): void
         $this->assertTrue($database->getDestination()->getDocument('testDeleteMirroredDocument', $document->getId())->isEmpty());
     }
 
-    public function testCreateDocumentsSkipDuplicatesDoesNotDivergeDestination(): void
+    public function testCreateDocumentsSkipDuplicatesBackfillsDestination(): void
     {
         $database = $this->getDatabase();
         $collection = 'mirrorSkipDup';
@@ -331,7 +331,9 @@ public function testCreateDocumentsSkipDuplicatesDoesNotDivergeDestination(): vo
         ], documentSecurity: false);
 
         // Seed the SOURCE only (bypass the mirror) with the row we want to
-        // skipDuplicates over later. Destination intentionally does NOT have it.
+        // skipDuplicates over later. Destination intentionally does NOT have it —
+        // this simulates an in-flight backfill where the collection is marked
+        // 'upgraded' (schema mirrored) but not every row has reached destination.
         $database->getSource()->createDocument($collection, new Document([
             '$id' => 'dup',
             'name' => 'Original',
@@ -341,7 +343,6 @@ public function testCreateDocumentsSkipDuplicatesDoesNotDivergeDestination(): vo
             ],
         ]));
 
-        // Sanity check setup
         $this->assertSame(
             'Original',
             $database->getSource()->getDocument($collection, 'dup')->getAttribute('name')
@@ -369,6 +370,7 @@ public function testCreateDocumentsSkipDuplicatesDoesNotDivergeDestination(): vo
             ]),
         ]));
 
+        // Source: INSERT IGNORE — 'dup' is a no-op, keeps 'Original'.
         $this->assertSame(
             'Original',
             $database->getSource()->getDocument($collection, 'dup')->getAttribute('name')
@@ -378,11 +380,13 @@ public function testCreateDocumentsSkipDuplicatesDoesNotDivergeDestination(): vo
             $database->getSource()->getDocument($collection, 'fresh')->getAttribute('name')
         );
 
-        // A source-skipped duplicate is a no-op and must not propagate to
-        // destination. Only the genuinely-inserted 'fresh' row should mirror.
-        $this->assertTrue(
-            $database->getDestination()->getDocument($collection, 'dup')->isEmpty(),
-            'Source-skipped doc must not be inserted into destination'
+        // Destination: 'dup' is NOT a duplicate there, so destination's own
+        // INSERT IGNORE inserts it. This prevents permanent divergence when
+        // destination is still catching up on rows that already exist on source.
+        $this->assertSame(
+            'WouldBe',
+            $database->getDestination()->getDocument($collection, 'dup')->getAttribute('name'),
+            'Source-skipped doc must still insert on destination when absent there'
         );
         $this->assertSame(
             'Fresh',