test(migrations): close the gaps the upgrade assertions left open

Three holes in the new suite, all of which let a test pass without proving
what its name claims:

- A migration recorded twice, once per replica, each with
  applied_steps_count = 1, slipped past both the step-count check and
  migration_names(), which collapses the history into a set. Reject
  duplicate migration_name rows outright.
- auth_traffic only asserted the failures it had seen by the time
  keep_serving hit its target. A request failing after that, or on the
  other replica while the test waited on one stream, was recorded and
  never read. Assert the recorded failures once the thread has joined.
- The rolling test warmed the baseline replica's virtual-key cache before
  the upgrade, and that cache holds for 60 seconds by default
  (UserAPIKeyCacheTTLEnum.in_memory_cache_ttl). The candidate migrates
  well inside that window, so the post-upgrade requests could be served
  from cache without ever repeating the whole-row token lookup that the
  stale prepared statement breaks. Drive the baseline replica with a key
  minted after the schema moved, which it has never seen and must resolve
  from the database.

Re-ran against v1.101.0 -> v1.102.0: 6 passed.

tests/e2e/migrations/test_rolling_upgrade.py

@@ -32,6 +32,8 @@ class TestRollingUpgrade:
                     ready((new,), baseline_database)
                     assert_upgraded(before, migration_names(baseline_database))
                     keep_serving(traffic, "the baseline replica authenticating after the schema moved")
+                    with auth_traffic(old, provision(new)[0]) as uncached:
+                        keep_serving(uncached, "the baseline replica resolving a key minted after the schema moved")
             assert_history_clean(baseline_database)
             assert CACHED_PLAN not in old.logs(), "The baseline replica hit a stale prepared statement"
             assert old.state().Running, "The baseline replica died during the upgrade"

tests/e2e/migrations/upgrade.py

@@ -88,6 +88,9 @@ def auth_traffic(replica: Replica, key: str, interval: float = 0.05) -> Generato
         stop.set()
         thread.join(30)
         assert not thread.is_alive(), "Auth traffic thread did not stop"
+    assert not outcomes.failures, (
+        f"Virtual-key auth failed on {replica.name} after the traffic window closed: {outcomes.failures[:5]}"
+    )
 
 
 def keep_serving(outcomes: Outcomes, description: str, calls: int = 20) -> int:
@@ -105,6 +108,10 @@ def assert_history_clean(database: Database) -> None:
     assert database.query(
         "SELECT count(*) FROM _prisma_migrations WHERE finished_at IS NULL OR rolled_back_at IS NOT NULL"
     ) == ((0,),), "The upgrade left an unfinished or rolled-back migration behind"
+    assert database.query(
+        "SELECT count(*) FROM (SELECT migration_name FROM _prisma_migrations GROUP BY migration_name "
+        "HAVING count(*) > 1) duplicated"
+    ) == ((0,),), "A migration was recorded more than once, so it ran on more than one replica"
 
 
 def assert_upgraded(before: frozenset[str], after: frozenset[str]) -> frozenset[str]: