feat(patch): delta schema 4 — register line_ref + line_dh in the patch/hash contract

.github/scripts/test_manual_release_workflow.py

@@ -148,17 +148,26 @@ def test_phase2_implementation_commit_is_part_of_release_identity(self):
             stage,
         )
 
-    def test_patch_fan_allows_only_the_supported_schema_transition(self):
+    def test_patch_fan_allows_only_the_supported_schema_transitions(self):
         patch_fan = self.step("Produce + verify patch fan")
 
-        self.assertIn('[ "$PREV_SCHEMA" = 2 ] && [ "$THIS_SCHEMA" = 3 ]', patch_fan)
+        supported = (
+            '[ "$PREV_SCHEMA" = 1 ] && [ "$THIS_SCHEMA" = 4 ]',
+            '[ "$PREV_SCHEMA" = 2 ] && [ "$THIS_SCHEMA" = 3 ]',
+            '[ "$PREV_SCHEMA" = 2 ] && [ "$THIS_SCHEMA" = 4 ]',
+            '[ "$PREV_SCHEMA" = 3 ] && [ "$THIS_SCHEMA" = 4 ]',
+        )
+        for transition in supported:
+            self.assertIn(transition, patch_fan)
+            self.assertLess(
+                patch_fan.index(transition),
+                patch_fan.index("gradle :generator-common:producePatchAndVerify"),
+            )
         self.assertIn("producing the supported cross-schema delta", patch_fan)
         self.assertIn("is unsupported — skip anchor", patch_fan)
         self.assertNotIn("cross-schema delta unsupported", patch_fan)
-        self.assertLess(
-            patch_fan.index('[ "$PREV_SCHEMA" = 2 ] && [ "$THIS_SCHEMA" = 3 ]'),
-            patch_fan.index("gradle :generator-common:producePatchAndVerify"),
-        )
+        self.assertNotIn('[ "$PREV_SCHEMA" = 1 ] && [ "$THIS_SCHEMA" = 2 ]', patch_fan)
+        self.assertNotIn('[ "$PREV_SCHEMA" = 1 ] && [ "$THIS_SCHEMA" = 3 ]', patch_fan)
 
     def test_pinned_sefaria_archive_uses_its_explicit_root_contract(self):
         extract = self.step("Verify pinned lineage and extract exact inputs")

.github/workflows/manual-generate-release.yml

@@ -1424,13 +1424,17 @@ jobs:
             PREV_DB=prev-dbs/seforim.db
             test -s "$PREV_DB" || { echo "::error::seforim.db.zst not found in release $TAG"; exit 1; }
 
-            # Schema 2 → 3 is an explicitly supported migration: the producer
-            # emits CREATE TABLE/INDEX migrations and verifies a real apply.
-            # Other unknown schema transitions remain fail-closed per anchor.
+            # Explicitly supported schema transitions. The producer derives
+            # contract promotions from both signed schema versions, rebuilds
+            # newly tracked tables from a full snapshot, and verifies a real
+            # apply. Unknown transitions remain fail-closed per anchor.
             PREV_SCHEMA=$(read_schema "$PREV_DB") || exit 1
             if [ "$PREV_SCHEMA" != "$THIS_SCHEMA" ]; then
-              if [ "$PREV_SCHEMA" = 2 ] && [ "$THIS_SCHEMA" = 3 ]; then
-                echo "schema 2 → 3 — producing the supported cross-schema delta"
+              if { [ "$PREV_SCHEMA" = 2 ] && [ "$THIS_SCHEMA" = 3 ]; } || \
+                 { [ "$PREV_SCHEMA" = 1 ] && [ "$THIS_SCHEMA" = 4 ]; } || \
+                 { [ "$PREV_SCHEMA" = 2 ] && [ "$THIS_SCHEMA" = 4 ]; } || \
+                 { [ "$PREV_SCHEMA" = 3 ] && [ "$THIS_SCHEMA" = 4 ]; }; then
+                echo "schema $PREV_SCHEMA → $THIS_SCHEMA — producing the supported cross-schema delta"
               else
                 echo "schema $PREV_SCHEMA → $THIS_SCHEMA is unsupported — skip anchor"
                 rm -rf prev-dbs

generator/common/src/jvmMain/kotlin/io/github/kdroidfilter/seforimlibrary/common/patch/LogicalContentHasher.kt

@@ -71,7 +71,7 @@ class LogicalContentHasher(
     }
 
     companion object {
-        /** Schema 1/2 hash contract. Never append future tables here. */
+        /** Schema-2 hash contract. Never append future tables here. */
         val TABLES_SCHEMA_2: List<String> = listOf(
             "source",
             "author",
@@ -109,17 +109,31 @@ class LogicalContentHasher(
             "schema_meta",
         )
 
+        /** Schema-1 hash contract predates the book_base_text junction. */
+        val TABLES_SCHEMA_1: List<String> = TABLES_SCHEMA_2.filterNot { it == "book_base_text" }
+
         /** Schema 3 adds the sparse per-side visibility table after link coverage. */
         val TABLES_SCHEMA_3: List<String> = TABLES_SCHEMA_2.toMutableList().apply {
             add(indexOf("link_coverage") + 1, "link_suppressed_side")
         }
 
+        /**
+         * Schema 4 adds the canonical line-reference index and the
+         * dibbur-hamatchil index right after line_toc.
+         */
+        val TABLES_SCHEMA_4: List<String> = TABLES_SCHEMA_3.toMutableList().apply {
+            add(indexOf("line_toc") + 1, "line_ref")
+            add(indexOf("line_ref") + 1, "line_dh")
+        }
+
         /** Current-schema default for build-time diagnostics and current DB tests. */
-        val DEFAULT_TABLES: List<String> = TABLES_SCHEMA_3
+        val DEFAULT_TABLES: List<String> = TABLES_SCHEMA_4
 
         fun tablesForSchemaVersion(schemaVersion: Int): List<String> = when (schemaVersion) {
-            1, 2 -> TABLES_SCHEMA_2
+            1 -> TABLES_SCHEMA_1
+            2 -> TABLES_SCHEMA_2
             3 -> TABLES_SCHEMA_3
+            4 -> TABLES_SCHEMA_4
             else -> error("Unsupported logical-hash schema version $schemaVersion")
         }
 

generator/common/src/jvmMain/kotlin/io/github/kdroidfilter/seforimlibrary/common/patch/PatchApplier.kt

@@ -96,17 +96,16 @@ class PatchApplier(
     }
 
     private fun runMigrations(conn: Connection): Int {
-        var count = 0
-        conn.createStatement().use { st ->
+        // Materialise the attached patch rows before executing DDL. Keeping a
+        // live sqlite_master-backed result set open while DROP/CREATE changes
+        // the main schema can produce SQLITE_LOCKED on SQLite JDBC.
+        val migrations = conn.createStatement().use { st ->
             st.executeQuery("SELECT sql FROM patch.migrations ORDER BY version ASC").use { rs ->
-                while (rs.next()) {
-                    val sql = rs.getString(1)
-                    conn.createStatement().use { it.execute(sql) }
-                    count++
-                }
+                buildList { while (rs.next()) add(rs.getString(1)) }
             }
         }
-        return count
+        for (sql in migrations) conn.createStatement().use { it.execute(sql) }
+        return migrations.size
     }
 
     private fun runUpserts(conn: Connection): Map<String, Int> {
@@ -169,7 +168,9 @@ class PatchApplier(
 
     /**
      * Reads `patch.patch_meta.schema_version` and refuses to apply a patch
-     * whose schema is newer than [PatchDbSchema.CURRENT_VERSION]. Without
+     * whose format version is outside the supported 1..[PatchDbSchema.CURRENT_VERSION]
+     * range. This is the patch artifact format, not the target DB schema from
+     * the release manifest. Without
      * this check, an older client could silently mis-apply a future-schema
      * patch.db (new tables ignored, new patch_meta keys not honoured),
      * producing a DB that "passed" the FK check but is semantically wrong.
@@ -180,10 +181,11 @@ class PatchApplier(
                 "patch.db is missing patch_meta.schema_version — refusing to apply " +
                     "(likely a corrupt or hand-built patch).",
             )
-        if (patchSchemaVersion > PatchDbSchema.CURRENT_VERSION) {
+        if (patchSchemaVersion !in 1..PatchDbSchema.CURRENT_VERSION) {
             throw IllegalStateException(
                 "patch.db carries schema_version=$patchSchemaVersion but this build " +
-                    "of the applier only understands up to ${PatchDbSchema.CURRENT_VERSION}. " +
+                    "of the applier only understands versions " +
+                    "1..${PatchDbSchema.CURRENT_VERSION}. " +
                     "Upgrade the client before applying this patch.",
             )
         }

generator/common/src/jvmMain/kotlin/io/github/kdroidfilter/seforimlibrary/common/patch/PatchDbProducer.kt

@@ -10,7 +10,7 @@ import java.sql.DriverManager
 /**
  * Produces a `patch.db` from two seforim.db snapshots (previous + current).
  *
- * For each table in [PATCH_TABLES_IN_FK_ORDER]:
+ * For each table in the signed `toSchemaVersion` contract:
  *   1. Reads the column list + PK from the **new** DB.
  *   2. Creates `upsert_<table>` mirroring those columns + PK.
  *   3. Inserts every row of `new.<table>` that either isn't in `prev.<table>`
@@ -19,6 +19,11 @@ import java.sql.DriverManager
  *      `(pk…)` tuple present in `prev` but missing from `new`.
  *
  * See `DELTA_UPDATE_PLAN.md` §6.6.
+ *
+ * A table promoted between schema contracts is rebuilt from the target DB's
+ * full snapshot even when it already exists physically in the previous full
+ * DB. Its unsigned physical state was never guaranteed to exist on clients
+ * that reached that release through an older-schema patch.
  */
 class PatchDbProducer(
     private val logger: Logger = Logger.withTag("PatchDbProducer"),
@@ -39,7 +44,17 @@ class PatchDbProducer(
         fromVersion: Int,
         toVersion: Int,
         migrations: List<Pair<Int, String>> = emptyList(),
+        fromSchemaVersion: Int = PatchDbSchema.CURRENT_VERSION,
+        toSchemaVersion: Int = PatchDbSchema.CURRENT_VERSION,
     ): Output {
+        require(fromSchemaVersion <= toSchemaVersion) {
+            "Schema downgrade $fromSchemaVersion -> $toSchemaVersion is not supported"
+        }
+        val fromTables = patchTablesForSchemaVersion(fromSchemaVersion)
+        val targetTables = patchTablesForSchemaVersion(toSchemaVersion)
+        val fromTableNames = fromTables.mapTo(HashSet()) { it.name }
+        val promotedTables = targetTables.mapTo(HashSet()) { it.name } - fromTableNames
+
         Files.createDirectories(outputPath.toAbsolutePath().parent)
         val tmp = outputPath.resolveSibling("${outputPath.fileName}.tmp")
         if (Files.exists(tmp)) Files.delete(tmp)
@@ -51,34 +66,54 @@ class PatchDbProducer(
         DriverManager.getConnection("jdbc:sqlite:${tmp.toAbsolutePath()}").use { conn ->
             conn.autoCommit = false
             applyBaseDdl(conn)
+            // patch_meta.schema_version describes the patch artifact format
+            // understood by PatchApplier. It is deliberately independent of
+            // the target DB's logical schema version carried by the release
+            // manifest (and may therefore be 4 for a 2 -> 3 DB transition).
             writeMetadata(conn, fromVersion, toVersion)
             attach(conn, "prev", prevDb)
             attach(conn, "new", newDb)
             val nextMigrationVersion = (migrations.maxOfOrNull { it.first } ?: 0) + 1
-            writeMigrations(conn, migrations + inferCreateTableMigrations(conn, nextMigrationVersion))
+            writeMigrations(
+                conn,
+                migrations + inferCreateTableMigrations(
+                    conn = conn,
+                    firstVersion = nextMigrationVersion,
+                    targetTables = targetTables,
+                    promotedTables = promotedTables,
+                ),
+            )
 
             // Materialise upsert_/delete_ tables based on the new DB's actual
             // schema. The producer is generic — every table in our config list
             // is processed identically.
-            for (table in PATCH_TABLES_IN_FK_ORDER) {
+            for (table in targetTables) {
                 if (!tableExists(conn, "new", table.name)) continue
                 PatchDbSchema.createUpsertTable(conn, "new", table)
                 PatchDbSchema.createDeleteTable(conn, "new", table)
             }
 
-            for (table in PATCH_TABLES_IN_FK_ORDER) {
-                upsertCounts[table.name] = scanUpserts(conn, table)
+            for (table in targetTables) {
+                upsertCounts[table.name] = scanUpserts(
+                    conn,
+                    table,
+                    forceFullSnapshot = table.name in promotedTables,
+                )
             }
-            for (table in PATCH_TABLES_IN_FK_ORDER) {
-                deleteCounts[table.name] = scanDeletes(conn, table)
+            for (table in targetTables) {
+                deleteCounts[table.name] = scanDeletes(
+                    conn,
+                    table,
+                    ignorePrevious = table.name in promotedTables,
+                )
             }
 
             // Fail fast on secondary-UNIQUE collisions: catches the case
             // where prev and new were generated from different build_state.db
             // lineages (e.g. same `topic.name` allocated under different ids),
             // which would otherwise blow up mid-transaction in the applier
             // with an opaque "UNIQUE constraint failed" error.
-            assertNoSecondaryUniqueCollisions(conn)
+            assertNoSecondaryUniqueCollisions(conn, targetTables)
 
             // Commit BEFORE detach so SQLite isn't holding locks on the
             // attached DBs through an open transaction.
@@ -126,12 +161,24 @@ class PatchDbProducer(
         }
     }
 
-    private fun inferCreateTableMigrations(conn: Connection, firstVersion: Int): List<Pair<Int, String>> {
+    private fun inferCreateTableMigrations(
+        conn: Connection,
+        firstVersion: Int,
+        targetTables: List<PatchTable>,
+        promotedTables: Set<String>,
+    ): List<Pair<Int, String>> {
         val out = ArrayList<Pair<Int, String>>()
         var version = firstVersion
-        for (table in PATCH_TABLES_IN_FK_ORDER) {
+        for (table in targetTables) {
             if (!tableExists(conn, "new", table.name)) continue
-            if (tableExists(conn, "prev", table.name)) continue
+            val promoted = table.name in promotedTables
+            if (!promoted && tableExists(conn, "prev", table.name)) continue
+
+            // A table that existed physically in a previous DB but was not in
+            // its signed schema contract may be absent (or carry arbitrary
+            // stale rows) on clients that reached that release by delta. Reset
+            // it and ship a full snapshot so both client shapes converge.
+            if (promoted) out += version++ to "DROP TABLE IF EXISTS \"${table.name}\""
 
             readCreateSql(conn, "new", "table", table.name)?.let { sql ->
                 out += version++ to sql
@@ -184,11 +231,11 @@ class PatchDbProducer(
         conn.createStatement().use { it.execute("DETACH DATABASE $alias") }
     }
 
-    private fun scanUpserts(conn: Connection, table: PatchTable): Int {
+    private fun scanUpserts(conn: Connection, table: PatchTable, forceFullSnapshot: Boolean = false): Int {
         val cols = PatchDbSchema.readTableInfo(conn, "new", table.name).map { it.name }
         if (cols.isEmpty()) return 0
         val colsCsv = cols.joinToString(",") { "\"$it\"" }
-        if (!tableExists(conn, "prev", table.name)) {
+        if (forceFullSnapshot || !tableExists(conn, "prev", table.name)) {
             val sql = """
                 INSERT INTO "upsert_${table.name}" ($colsCsv)
                 SELECT $colsCsv
@@ -218,8 +265,9 @@ class PatchDbProducer(
         return conn.createStatement().use { it.executeUpdate(sql) }
     }
 
-    private fun scanDeletes(conn: Connection, table: PatchTable): Int {
+    private fun scanDeletes(conn: Connection, table: PatchTable, ignorePrevious: Boolean = false): Int {
         if (table.primaryKey.isEmpty()) return 0
+        if (ignorePrevious) return 0
         if (!tableExists(conn, "prev", table.name)) return 0
         val pkCsv = table.primaryKey.joinToString(",") { "\"$it\"" }
         val joinCond = table.primaryKey.joinToString(" AND ") { "new.\"$it\" = prev.\"$it\"" }
@@ -245,8 +293,8 @@ class PatchDbProducer(
      * surfacing them here gives the operator a clear, actionable error
      * instead of a mid-transaction crash in [PatchApplier].
      */
-    private fun assertNoSecondaryUniqueCollisions(conn: Connection) {
-        for (table in PATCH_TABLES_IN_FK_ORDER) {
+    private fun assertNoSecondaryUniqueCollisions(conn: Connection, tables: List<PatchTable>) {
+        for (table in tables) {
             if (!tableExists(conn, "new", table.name)) continue
             if (!tableExists(conn, "prev", table.name)) continue
             if (!tableExists(conn, "main", "upsert_${table.name}")) continue

generator/common/src/jvmMain/kotlin/io/github/kdroidfilter/seforimlibrary/common/patch/PatchDbSchema.kt

@@ -19,7 +19,7 @@ import java.sql.Connection
  */
 internal object PatchDbSchema {
 
-    const val CURRENT_VERSION: Int = 3
+    const val CURRENT_VERSION: Int = 4
 
     /** Fixed-shape tables (metadata + auxiliaries). */
     val baseStatements: List<String> = listOf(

generator/common/src/jvmMain/kotlin/io/github/kdroidfilter/seforimlibrary/common/patch/PatchPipelineCli.kt

@@ -53,6 +53,8 @@ fun main(args: Array<String>) {
         outputPath = outPath,
         fromVersion = from,
         toVersion = to,
+        fromSchemaVersion = fromSchemaVersion,
+        toSchemaVersion = toSchemaVersion,
     )
     val totalUpserts = output.upsertCounts.values.sum()
     val totalDeletes = output.deleteCounts.values.sum()
@@ -74,7 +76,12 @@ fun main(args: Array<String>) {
         // exactly across the v13→v14/v14→v15 verifications. HARD gate: a
         // patch that does not reproduce the target byte-for-logical-byte is
         // a broken distribution artifact and must never ship with a warning.
-        PatchApplier(logger).apply(conn = conn, patchDb = outPath)
+        PatchApplier(logger).apply(
+            conn = conn,
+            patchDb = outPath,
+            expectedToContentHash = newHash,
+            expectedToSchemaVersion = toSchemaVersion,
+        )
         val appliedHash = LogicalContentHasher.forSchemaVersion(toSchemaVersion).compute(conn)
         check(appliedHash == newHash) {
             "Patch verification FAILED: applied=$appliedHash expected=$newHash — " +

generator/common/src/jvmMain/kotlin/io/github/kdroidfilter/seforimlibrary/common/patch/PatchTables.kt

@@ -59,6 +59,11 @@ internal val PATCH_TABLES_IN_FK_ORDER: List<PatchTable> = listOf(
     PatchTable("tocEntry",           listOf("id"),       updatable = true),
     PatchTable("line",               listOf("id"),       updatable = true),
     PatchTable("line_toc",           listOf("lineId"),   updatable = true),
+    // Schema 4. Canonical line-reference index — pure key table (PK == all
+    // columns), so there is nothing to update on conflict.
+    PatchTable("line_ref",           listOf("bookId", "refKeyHash", "lineIndex"), updatable = false),
+    // Schema 4. Dibbur-hamatchil index — pure key table, same shape.
+    PatchTable("line_dh",            listOf("bookId", "dhText", "lineIndex"), updatable = false),
 
     // Links.
     PatchTable("link",               listOf("id"),       updatable = true),
@@ -86,6 +91,23 @@ internal val PATCH_TABLES_IN_FK_ORDER: List<PatchTable> = listOf(
     PatchTable("schema_meta",        listOf("key"),      updatable = true),
 )
 
+/** Schema-3 contract retained for updater compatibility tests. */
+internal val PATCH_TABLES_SCHEMA_3: List<PatchTable> =
+    PATCH_TABLES_IN_FK_ORDER.filterNot { it.name in setOf("line_ref", "line_dh") }
+
 /** Schema-2 contract retained for updater compatibility tests. */
 internal val PATCH_TABLES_SCHEMA_2: List<PatchTable> =
-    PATCH_TABLES_IN_FK_ORDER.filterNot { it.name == "link_suppressed_side" }
+    PATCH_TABLES_SCHEMA_3.filterNot { it.name == "link_suppressed_side" }
+
+/** Schema-1 contract predates the book_base_text junction. */
+internal val PATCH_TABLES_SCHEMA_1: List<PatchTable> =
+    PATCH_TABLES_SCHEMA_2.filterNot { it.name == "book_base_text" }
+
+/** Exact patch-table contract used to produce a database schema version. */
+internal fun patchTablesForSchemaVersion(schemaVersion: Int): List<PatchTable> = when (schemaVersion) {
+    1 -> PATCH_TABLES_SCHEMA_1
+    2 -> PATCH_TABLES_SCHEMA_2
+    3 -> PATCH_TABLES_SCHEMA_3
+    4 -> PATCH_TABLES_IN_FK_ORDER
+    else -> error("Unsupported patch-table schema version $schemaVersion")
+}