diff --git a/installer/database/migrations/migrate_0.7.31.sql b/installer/database/migrations/migrate_0.7.31.sql
index 43950edc..0f0398d1 100644
--- a/installer/database/migrations/migrate_0.7.31.sql
+++ b/installer/database/migrations/migrate_0.7.31.sql
@@ -112,6 +112,13 @@ SET @idx_exists = (SELECT COUNT(*) FROM information_schema.STATISTICS WHERE TABL
 SET @sql = IF(@idx_exists = 0, "ALTER TABLE `libri` ADD KEY `idx_lt_rating` (`rating`)", "SELECT 1");
 PREPARE _s FROM @sql; EXECUTE _s; DEALLOCATE PREPARE _s;
 
+-- Normalise any out-of-range rating BEFORE adding the CHECK, otherwise the
+-- ALTER ... ADD CONSTRAINT fails on pre-existing rows (e.g. a legacy import that
+-- wrote 0 or a 1-10 scale) and aborts the whole migration. Idempotent: on an
+-- install that already has the constraint there are no violating rows, so this
+-- touches nothing. Keep NULLs as-is (they satisfy the constraint).
+UPDATE `libri` SET `rating` = NULL WHERE `rating` IS NOT NULL AND `rating` NOT BETWEEN 1 AND 5;
+
 -- CHECK constraint names are schema-unique (not per-table) and this MySQL/
 -- MariaDB exposes them in CHECK_CONSTRAINTS (keyed by schema+name), NOT
 -- TABLE_CONSTRAINTS — probing the wrong table would re-issue the ADD on an
diff --git a/storage/plugins/book-club/src/Repo.php b/storage/plugins/book-club/src/Repo.php
index 32c7582f..a39682ea 100644
--- a/storage/plugins/book-club/src/Repo.php
+++ b/storage/plugins/book-club/src/Repo.php
@@ -14,6 +14,14 @@
  */
 class Repo
 {
+    /**
+     * `editori.nome` intentionally has no UNIQUE constraint because the core
+     * catalogue permits homonyms. Serialize Book Club publisher resolution on
+     * the database server instead, so separate PHP workers share the same lock.
+     */
+    private const PUBLISHER_LOCK_NAME = 'pinakes:bookclub:publisher';
+    private const PUBLISHER_LOCK_TIMEOUT_SECONDS = 10;
+
     private mysqli $db;
 
     public function __construct(mysqli $db)
@@ -159,35 +167,106 @@ private function findOrCreatePublisher(?string $name): ?int
             return null;
         }
 
-        $stmt = $this->db->prepare('SELECT id FROM editori WHERE nome = ? ORDER BY id ASC LIMIT 1');
+        $this->acquirePublisherLock();
+        try {
+            $sel = $this->db->prepare('SELECT id FROM editori WHERE nome = ? ORDER BY id ASC LIMIT 1');
+            if ($sel === false) {
+                throw new \RuntimeException('publisher lookup prepare failed');
+            }
+            $sel->bind_param('s', $name);
+            if (!$sel->execute()) {
+                $err = $sel->error;
+                $sel->close();
+                throw new \RuntimeException('publisher lookup failed: ' . $err);
+            }
+            $selResult = $sel->get_result();
+            $sel->close();
+            if ($selResult === false) {
+                throw new \RuntimeException('publisher lookup get_result failed');
+            }
+            $row = $selResult->fetch_assoc();
+            if ($row !== null) {
+                return (int) $row['id'];
+            }
+
+            $ins = $this->db->prepare('INSERT INTO editori (nome) VALUES (?)');
+            if ($ins === false) {
+                throw new \RuntimeException('publisher insert prepare failed');
+            }
+            $ins->bind_param('s', $name);
+            if (!$ins->execute()) {
+                $err = $ins->error;
+                $ins->close();
+                throw new \RuntimeException('publisher insert failed: ' . $err);
+            }
+            $publisherId = $ins->insert_id;
+            $ins->close();
+
+            if ($publisherId <= 0) {
+                throw new \RuntimeException('publisher insert returned no id');
+            }
+            return $publisherId;
+        } finally {
+            $this->releasePublisherLock();
+        }
+    }
+
+    private function acquirePublisherLock(): void
+    {
+        $lockName = self::PUBLISHER_LOCK_NAME;
+        $timeout = self::PUBLISHER_LOCK_TIMEOUT_SECONDS;
+        // GET_LOCK is server-wide, not per-database — scope the name to the
+        // current schema (computed server-side via CONCAT + DATABASE()) so
+        // independent Pinakes installs sharing one MySQL server don't serialize
+        // each other's imports. Mirrors ContributorBackfill's lock naming.
+        $stmt = $this->db->prepare("SELECT GET_LOCK(CONCAT(?, ':', DATABASE()), ?) AS acquired");
         if ($stmt === false) {
-            throw new \RuntimeException('publisher lookup prepare failed');
+            throw new \RuntimeException('publisher lock prepare failed');
         }
-        $stmt->bind_param('s', $name);
+        $stmt->bind_param('si', $lockName, $timeout);
         if (!$stmt->execute()) {
             $err = $stmt->error;
             $stmt->close();
-            throw new \RuntimeException('publisher lookup failed: ' . $err);
+            throw new \RuntimeException('publisher lock failed: ' . $err);
         }
-        $row = $stmt->get_result()->fetch_assoc();
+        $result = $stmt->get_result();
         $stmt->close();
-        if ($row !== null) {
-            return (int) $row['id'];
+        if ($result === false) {
+            throw new \RuntimeException('publisher lock get_result failed');
         }
+        $row = $result->fetch_assoc();
+        if ((int) ($row['acquired'] ?? 0) !== 1) {
+            throw new \RuntimeException('publisher lock timed out');
+        }
+    }
 
-        $stmt = $this->db->prepare('INSERT INTO editori (nome) VALUES (?)');
+    private function releasePublisherLock(): void
+    {
+        $lockName = self::PUBLISHER_LOCK_NAME;
+        $stmt = $this->db->prepare("SELECT RELEASE_LOCK(CONCAT(?, ':', DATABASE())) AS released");
         if ($stmt === false) {
-            throw new \RuntimeException('publisher insert prepare failed');
+            SecureLogger::error('[BookClub] publisher lock release prepare failed: ' . $this->db->error);
+            return;
         }
-        $stmt->bind_param('s', $name);
+        $stmt->bind_param('s', $lockName);
         if (!$stmt->execute()) {
-            $err = $stmt->error;
+            SecureLogger::error('[BookClub] publisher lock release failed: ' . $stmt->error);
             $stmt->close();
-            throw new \RuntimeException('publisher insert failed: ' . $err);
+            return;
         }
-        $id = (int) $stmt->insert_id;
+        // Guard get_result() === false here especially: this runs inside the
+        // finally of findOrCreatePublisher(), so an uncaught Error would mask a
+        // propagating exception — the very thing this lock protocol avoids.
+        $result = $stmt->get_result();
         $stmt->close();
-        return $id;
+        if ($result === false) {
+            SecureLogger::error('[BookClub] publisher lock release get_result failed: ' . $this->db->error);
+            return;
+        }
+        $row = $result->fetch_assoc();
+        if ((int) ($row['released'] ?? 0) !== 1) {
+            SecureLogger::error('[BookClub] publisher lock was not owned by this connection');
+        }
     }
 
     private function attachPrimaryPublisher(int $libroId, ?int $publisherId): void
diff --git a/tests/bookclub-publisher-dedup.unit.php b/tests/bookclub-publisher-dedup.unit.php
new file mode 100644
index 00000000..39fb03ae
--- /dev/null
+++ b/tests/bookclub-publisher-dedup.unit.php
@@ -0,0 +1,202 @@
+<?php
+declare(strict_types=1);
+
+/**
+ * Behavioural guard for the book-club Repo::findOrCreatePublisher() race fix.
+ *
+ * `editori.nome` intentionally has no UNIQUE constraint, so a NOT EXISTS insert
+ * can either duplicate or deadlock under real concurrency. The repository now
+ * serializes publisher resolution with a MySQL advisory lock shared by separate
+ * PHP/database connections. This test pins both the ordinary reuse contract and
+ * the overlapping two-connection path against a sandbox table shaped like
+ * `editori`.
+ *
+ * Runs against the LIVE local MySQL; touches only the zz_bc_editori table.
+ *
+ * Run:  php tests/bookclub-publisher-dedup.unit.php
+ */
+
+$root = dirname(__DIR__);
+mysqli_report(MYSQLI_REPORT_ERROR | MYSQLI_REPORT_STRICT);
+
+$env = [];
+foreach (preg_split('/\r?\n/', (string) @file_get_contents($root . '/.env')) as $line) {
+    $line = trim($line);
+    if ($line === '' || $line[0] === '#' || !str_contains($line, '=')) {
+        continue;
+    }
+    [$k, $v] = explode('=', $line, 2);
+    $k = trim($k);
+    $v = trim($v);
+    if (strlen($v) >= 2 && ($v[0] === '"' || $v[0] === "'") && $v[-1] === $v[0]) {
+        $v = substr($v, 1, -1);
+    }
+    $env[$k] = $v;
+}
+$socket = getenv('E2E_DB_SOCKET') ?: ($env['DB_SOCKET'] ?? '/opt/homebrew/var/mysql/mysql.sock');
+try {
+    $db = ($socket !== '' && file_exists($socket))
+        ? new mysqli(null, $env['DB_USER'] ?? '', $env['DB_PASS'] ?? ($env['DB_PASSWORD'] ?? ''), $env['DB_NAME'] ?? '', 0, $socket)
+        : new mysqli($env['DB_HOST'] ?? '127.0.0.1', $env['DB_USER'] ?? '', $env['DB_PASS'] ?? ($env['DB_PASSWORD'] ?? ''), $env['DB_NAME'] ?? '', (int) ($env['DB_PORT'] ?? 3306));
+} catch (\Throwable $e) {
+    echo "SKIP: database not reachable (" . $e->getMessage() . ")\n";
+    exit(0);
+}
+$db->set_charset('utf8mb4');
+
+try {
+    $db2 = ($socket !== '' && file_exists($socket))
+        ? new mysqli(null, $env['DB_USER'] ?? '', $env['DB_PASS'] ?? ($env['DB_PASSWORD'] ?? ''), $env['DB_NAME'] ?? '', 0, $socket)
+        : new mysqli($env['DB_HOST'] ?? '127.0.0.1', $env['DB_USER'] ?? '', $env['DB_PASS'] ?? ($env['DB_PASSWORD'] ?? ''), $env['DB_NAME'] ?? '', (int) ($env['DB_PORT'] ?? 3306));
+    $db2->set_charset('utf8mb4');
+} catch (\Throwable $e) {
+    $db->close();
+    echo "SKIP: second database connection not reachable (" . $e->getMessage() . ")\n";
+    exit(0);
+}
+
+const T = 'zz_bc_editori';
+$cleanup = static fn () => $db->query('DROP TABLE IF EXISTS `' . T . '`');
+$lockName = 'pinakes:test:bc-publisher:' . bin2hex(random_bytes(6));
+
+$pass = 0;
+$fail = 0;
+$check = static function (bool $ok, string $label) use (&$pass, &$fail): void {
+    if ($ok) { $pass++; echo "  OK  {$label}\n"; }
+    else     { $fail++; echo "  FAIL {$label}\n"; }
+};
+
+/** The exact locked lookup/insert contract Repo::findOrCreatePublisher() uses. */
+$findOrCreateWhileLocked = static function (mysqli $connection, string $name): int {
+    $sel = $connection->prepare('SELECT id FROM `' . T . '` WHERE nome = ? ORDER BY id ASC LIMIT 1');
+    $sel->bind_param('s', $name);
+    $sel->execute();
+    $row = $sel->get_result()->fetch_assoc();
+    $sel->close();
+    if ($row !== null) {
+        return (int) $row['id'];
+    }
+
+    $ins = $connection->prepare('INSERT INTO `' . T . '` (nome) VALUES (?)');
+    $ins->bind_param('s', $name);
+    $ins->execute();
+    $id = $ins->insert_id;
+    $ins->close();
+    return $id;
+};
+$acquireLock = static function (mysqli $connection, int $timeout = 5) use ($lockName): bool {
+    $stmt = $connection->prepare('SELECT GET_LOCK(?, ?) AS acquired');
+    $stmt->bind_param('si', $lockName, $timeout);
+    $stmt->execute();
+    $row = $stmt->get_result()->fetch_assoc();
+    $stmt->close();
+    return (int) ($row['acquired'] ?? 0) === 1;
+};
+$releaseLock = static function (mysqli $connection) use ($lockName): bool {
+    $stmt = $connection->prepare('SELECT RELEASE_LOCK(?) AS released');
+    $stmt->bind_param('s', $lockName);
+    $stmt->execute();
+    $row = $stmt->get_result()->fetch_assoc();
+    $stmt->close();
+    return (int) ($row['released'] ?? 0) === 1;
+};
+$findOrCreate = static function (mysqli $connection, string $name) use ($acquireLock, $releaseLock, $findOrCreateWhileLocked): int {
+    if (!$acquireLock($connection)) {
+        throw new RuntimeException('test publisher lock timed out');
+    }
+    try {
+        return $findOrCreateWhileLocked($connection, $name);
+    } finally {
+        $releaseLock($connection);
+    }
+};
+$count = static function (mysqli $connection, string $name): int {
+    $stmt = $connection->prepare('SELECT COUNT(*) FROM `' . T . '` WHERE nome = ?');
+    $stmt->bind_param('s', $name);
+    $stmt->execute();
+    $n = (int) $stmt->get_result()->fetch_row()[0];
+    $stmt->close();
+    return $n;
+};
+
+try {
+    $cleanup();
+    $db->query('CREATE TABLE `' . T . '` (
+        id INT AUTO_INCREMENT PRIMARY KEY,
+        nome VARCHAR(255) NOT NULL
+     ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci');
+
+    echo "A. Insert-if-absent + stable id\n";
+    $id1 = $findOrCreate($db, 'Einaudi');
+    $check($id1 > 0, "first call creates the publisher and returns its id ({$id1})");
+    $check($count($db, 'Einaudi') === 1, 'exactly one row exists after the first call');
+
+    $id2 = $findOrCreate($db, 'Einaudi');
+    $check($id2 === $id1, 'second call for the same name returns the SAME id (no create)');
+    $check($count($db, 'Einaudi') === 1, 'still exactly one row — no duplicate created');
+
+    echo "B. Distinct names get distinct rows\n";
+    $id3 = $findOrCreate($db, 'Adelphi');
+    $check($id3 !== $id1, "a different name creates a distinct row ({$id3})");
+    $check((int) $db->query('SELECT COUNT(*) FROM `' . T . '`')->fetch_row()[0] === 2, 'two rows total for two distinct names');
+
+    echo "C. Pre-existing row is reused, never re-inserted\n";
+    // Simulate a legacy duplicate already present (core tables allow homonyms):
+    $db->query("INSERT INTO `" . T . "` (nome) VALUES ('Adelphi')");
+    $dupCountBefore = $count($db, 'Adelphi');
+    $check($dupCountBefore === 2, 'a legacy duplicate can pre-exist (2 rows named Adelphi)');
+    $idA = $findOrCreate($db, 'Adelphi');
+    $check($count($db, 'Adelphi') === 2, 'find-or-create does NOT add a third row when the name already exists');
+    $check($idA === $id3, 'it returns the LOWEST existing id deterministically');
+
+    echo "D. Overlapping connections serialize instead of racing\n";
+    $check($acquireLock($db), 'first connection acquires the shared publisher lock');
+    $escapedLockName = $db2->real_escape_string($lockName);
+    $db2->query("SELECT GET_LOCK('{$escapedLockName}', 5) AS acquired", MYSQLI_ASYNC);
+    usleep(100_000);
+    $read = [$db2];
+    $error = [$db2];
+    $reject = [];
+    $check(mysqli_poll($read, $error, $reject, 0, 0) === 0, 'second connection waits while the first owns the lock');
+
+    $idConcurrent1 = $findOrCreateWhileLocked($db, 'Mondadori');
+    $check($releaseLock($db), 'first connection releases the lock after inserting');
+
+    $read = [$db2];
+    $error = [$db2];
+    $reject = [];
+    $check(mysqli_poll($read, $error, $reject, 5) === 1, 'second connection resumes after release');
+    $result = $db2->reap_async_query();
+    if (!$result instanceof mysqli_result) {
+        throw new RuntimeException('failed to reap asynchronous lock query');
+    }
+    $lockRow = $result->fetch_assoc();
+    $result->free();
+    $check((int) ($lockRow['acquired'] ?? 0) === 1, 'second connection acquires the same lock');
+    $idConcurrent2 = $findOrCreateWhileLocked($db2, 'Mondadori');
+    $check($releaseLock($db2), 'second connection releases the lock');
+    $check($idConcurrent2 === $idConcurrent1, 'both overlapping imports resolve the SAME publisher id');
+    $check($count($db, 'Mondadori') === 1, 'overlap creates exactly one publisher row');
+
+    // findOrCreatePublisher() operates on the CORE `editori` table via hardcoded
+    // SQL, so it can't be driven directly here without polluting real data — the
+    // protocol above is proven on a sandbox replica. Bind the sandbox proof to
+    // production by asserting the real code keeps the exact protocol invariants:
+    // acquire/release AND the per-database scoping (GET_LOCK is server-wide, so a
+    // regression that drops DATABASE() would reintroduce cross-tenant contention).
+    $repoSource = (string) file_get_contents($root . '/storage/plugins/book-club/src/Repo.php');
+    $check(str_contains($repoSource, 'GET_LOCK(CONCAT(') && str_contains($repoSource, 'DATABASE()'),
+        'production Repo scopes the advisory lock per database (GET_LOCK(CONCAT(?, DATABASE())))');
+    $check(str_contains($repoSource, 'RELEASE_LOCK(CONCAT('),
+        'production Repo releases the same per-database-scoped lock');
+} finally {
+    // Advisory locks are connection-scoped; make cleanup idempotent after a failed assertion/query.
+    try { $releaseLock($db); } catch (Throwable) {}
+    try { $releaseLock($db2); } catch (Throwable) {}
+    $cleanup();
+    $db2->close();
+    $db->close();
+}
+
+echo "\n" . ($fail === 0 ? "ALL {$pass} PASS\n" : "{$pass} PASS, {$fail} FAIL\n");
+exit($fail === 0 ? 0 : 1);
diff --git a/tests/migration-0.7.31-rating-cleanup.unit.php b/tests/migration-0.7.31-rating-cleanup.unit.php
new file mode 100644
index 00000000..25eb3fcd
--- /dev/null
+++ b/tests/migration-0.7.31-rating-cleanup.unit.php
@@ -0,0 +1,124 @@
+<?php
+declare(strict_types=1);
+
+/**
+ * Behavioural guard for the rating normalisation added to migrate_0.7.31.sql.
+ *
+ * The migration re-adds `chk_lt_rating` (CHECK rating IS NULL OR BETWEEN 1 AND 5)
+ * on `libri`. On an install that already has the `rating` column carrying legacy
+ * out-of-range values (e.g. a 0 = "unrated" or a 1-10 import), MySQL rejects the
+ * `ALTER ... ADD CONSTRAINT` and aborts the whole migration. The fix normalises
+ * those rows to NULL BEFORE the ADD:
+ *
+ *   UPDATE `libri` SET `rating` = NULL WHERE `rating` IS NOT NULL AND `rating` NOT BETWEEN 1 AND 5;
+ *
+ * This test reproduces the failure and proves the fix against a sandbox table
+ * shaped like the relevant part of `libri` (rating TINYINT UNSIGNED, no CHECK):
+ *   1. Adding the CHECK with bad rows present FAILS (proves the fix is needed).
+ *   2. The normalising UPDATE nulls only out-of-range values (keeps 1-5 + NULL).
+ *   3. After it, the CHECK ADD SUCCEEDS.
+ *   4. Re-running the UPDATE is a no-op (idempotent).
+ * It also asserts the real migration file actually contains the UPDATE before the
+ * CHECK ADD, so the behaviour can't drift out of the shipped SQL.
+ *
+ * Runs against the LIVE local MySQL; touches only the zz_rating_* table.
+ *
+ * Run:  php tests/migration-0.7.31-rating-cleanup.unit.php
+ */
+
+$root = dirname(__DIR__);
+mysqli_report(MYSQLI_REPORT_ERROR | MYSQLI_REPORT_STRICT);
+
+$env = [];
+foreach (preg_split('/\r?\n/', (string) @file_get_contents($root . '/.env')) as $line) {
+    $line = trim($line);
+    if ($line === '' || $line[0] === '#' || !str_contains($line, '=')) {
+        continue;
+    }
+    [$k, $v] = explode('=', $line, 2);
+    $k = trim($k);
+    $v = trim($v);
+    if (strlen($v) >= 2 && ($v[0] === '"' || $v[0] === "'") && $v[-1] === $v[0]) {
+        $v = substr($v, 1, -1);
+    }
+    $env[$k] = $v;
+}
+$socket = getenv('E2E_DB_SOCKET') ?: ($env['DB_SOCKET'] ?? '/opt/homebrew/var/mysql/mysql.sock');
+try {
+    $db = ($socket !== '' && file_exists($socket))
+        ? new mysqli(null, $env['DB_USER'] ?? '', $env['DB_PASS'] ?? ($env['DB_PASSWORD'] ?? ''), $env['DB_NAME'] ?? '', 0, $socket)
+        : new mysqli($env['DB_HOST'] ?? '127.0.0.1', $env['DB_USER'] ?? '', $env['DB_PASS'] ?? ($env['DB_PASSWORD'] ?? ''), $env['DB_NAME'] ?? '', (int) ($env['DB_PORT'] ?? 3306));
+} catch (\Throwable $e) {
+    echo "SKIP: database not reachable (" . $e->getMessage() . ")\n";
+    exit(0);
+}
+$db->set_charset('utf8mb4');
+
+const T = 'zz_rating_libri';
+$cleanup = static fn () => $db->query('DROP TABLE IF EXISTS `' . T . '`');
+
+$pass = 0;
+$fail = 0;
+$check = static function (bool $ok, string $label) use (&$pass, &$fail): void {
+    if ($ok) { $pass++; echo "  OK  {$label}\n"; }
+    else     { $fail++; echo "  FAIL {$label}\n"; }
+};
+$scalar = static function (string $sql) use ($db) {
+    $r = $db->query($sql);
+    $row = $r ? $r->fetch_row() : null;
+    return $row ? $row[0] : null;
+};
+$addCheckThrows = static function () use ($db): bool {
+    try {
+        $db->query('ALTER TABLE `' . T . '` ADD CONSTRAINT `chk_rt` '
+            . 'CHECK (`rating` IS NULL OR `rating` BETWEEN 1 AND 5)');
+        return false; // succeeded
+    } catch (\Throwable $e) {
+        return true;  // rejected
+    }
+};
+
+try {
+    // ── Static: the shipped migration normalises rating before adding the CHECK ──
+    echo "A. Static — migration file normalises before the CHECK ADD\n";
+    $sql = (string) file_get_contents($root . '/installer/database/migrations/migrate_0.7.31.sql');
+    $updatePos = strpos($sql, "UPDATE `libri` SET `rating` = NULL WHERE `rating` IS NOT NULL AND `rating` NOT BETWEEN 1 AND 5");
+    $chkPos = strpos($sql, "ADD CONSTRAINT `chk_lt_rating`");
+    $check($updatePos !== false, 'migrate_0.7.31.sql contains the rating-normalising UPDATE');
+    $check($updatePos !== false && $chkPos !== false && $updatePos < $chkPos,
+        'the UPDATE precedes the chk_lt_rating ADD (so bad rows are cleaned first)');
+
+    // ── Behavioural: reproduce the failure, then prove the fix ──
+    echo "B. Behavioural — bad data blocks the CHECK; the UPDATE unblocks it\n";
+    $cleanup();
+    $db->query('CREATE TABLE `' . T . '` (
+        id INT AUTO_INCREMENT PRIMARY KEY,
+        rating TINYINT UNSIGNED NULL
+     ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4');
+    // 3 = valid, 0 and 7 = out of range, NULL = valid (unrated).
+    $db->query('INSERT INTO `' . T . '` (rating) VALUES (3), (0), (7), (NULL)');
+
+    $check($addCheckThrows(), 'adding chk_lt_rating with out-of-range rows FAILS (bug reproduced)');
+
+    // The fix: normalise out-of-range values (the exact statement from the migration).
+    $db->query('UPDATE `' . T . '` SET `rating` = NULL WHERE `rating` IS NOT NULL AND `rating` NOT BETWEEN 1 AND 5');
+
+    $check((int) $scalar('SELECT COUNT(*) FROM `' . T . '` WHERE rating = 3') === 1, 'valid rating 3 is preserved');
+    $check((int) $scalar('SELECT COUNT(*) FROM `' . T . '` WHERE rating IS NULL') === 3, 'the two out-of-range rows joined the original NULL (3 NULLs)');
+    $check((int) $scalar('SELECT COUNT(*) FROM `' . T . '` WHERE rating IS NOT NULL AND rating NOT BETWEEN 1 AND 5') === 0, 'no out-of-range value remains');
+
+    $check(!$addCheckThrows(), 'after the UPDATE, adding chk_lt_rating SUCCEEDS');
+
+    // ── Idempotency: the UPDATE is a no-op on already-clean data ──
+    echo "C. Idempotent — re-running the UPDATE touches nothing\n";
+    $db->query('UPDATE `' . T . '` SET `rating` = NULL WHERE `rating` IS NOT NULL AND `rating` NOT BETWEEN 1 AND 5');
+    $affected = $db->affected_rows;
+    $check($affected === 0, 're-running the normalising UPDATE affects 0 rows');
+    $check((int) $scalar('SELECT COUNT(*) FROM `' . T . '` WHERE rating = 3') === 1, 'the valid rating still survives the second run');
+} finally {
+    $cleanup();
+    $db->close();
+}
+
+echo "\n" . ($fail === 0 ? "ALL {$pass} PASS\n" : "{$pass} PASS, {$fail} FAIL\n");
+exit($fail === 0 ? 0 : 1);