Refuse a torn export.pdb instead of converting it (G1d)

CHANGELOG.md

@@ -8,6 +8,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Fixed
+- **G1d — refuse a torn `export.pdb` instead of converting it.** A conversion on
+  a real stick published two playlist entries that the settled `export.pdb` does
+  not contain. Forensics on the drive established what happened: rekordbox last
+  wrote the pdb at 12:09:38 UTC, `convert` started 14 seconds later at 12:09:52,
+  and the file has not been modified since — so `convert` and the `verify` that
+  caught it read the *same* file, and the difference arose while reading a pdb
+  that was still settling. A stale slot read as present parses cleanly, lands in
+  a coherent chain, and is indistinguishable downstream from a real entry.
+
+  The page header already carries the contradiction: it declares `num_rows`, and
+  the reader decoded that field and threw it away. The parser now checks that the
+  present-bit count matches `num_rows`, and that every row offset points into the
+  heap between the page header and the backward-growing row index. Both invariants
+  hold on all 997 data pages of a real 3,673-track export, and ordinary deleted
+  slots — 104 of them in that same file — still parse normally.
+
+  Scope, stated plainly: this catches the demonstrated signature class. A torn
+  image whose header was written before the pages it describes satisfies both
+  checks and would still pass.
 - `verify` paired each source playlist with the wrong Engine list in three ways,
   every one of which invents discrepancies on a correct conversion — the failure
   mode that trains you to ignore verify. Same-named playlists in different

src/rb2engine/reader/pdb.py

@@ -213,32 +213,75 @@ def _is_data_page(page_flags: int) -> bool:
     return (page_flags & 0x40) == 0
 
 
+# num_rows is 11 bits; a page holding more present rows than that cannot state
+# its own count, so the cross-check below is skipped rather than mis-fired.
+_NUM_ROWS_MAX = 0x7FF
+
+
 def _iter_present_row_bases(
-    page: bytes, len_page: int, num_row_offsets: int
+    page: bytes,
+    len_page: int,
+    num_row_offsets: int,
+    num_rows: int = -1,
+    page_index: int = -1,
 ) -> list[int]:
-    """Absolute offsets within `page` of each *present* row body."""
+    """Absolute offsets within `page` of each *present* row body.
+
+    G1d — torn/mid-export page images. The row index grows backward from the
+    end of the page, so a row body lives strictly between the page header and
+    the index. Two structural facts are checked here rather than assumed:
+
+    * every present offset points into that heap region, and
+    * the number of present bits equals the ``num_rows`` the header declares.
+
+    Both hold on all 997 data pages of a real 3,673-track export. The second
+    exists because a conversion once published playlist entries the settled
+    ``export.pdb`` does not contain: a stale slot read as present parses
+    cleanly, lands in a coherent chain, and is indistinguishable downstream
+    from a real entry. The page header already carries the count that
+    contradicts it; the walker simply discarded it.
+    """
     if num_row_offsets <= 0:
         return []
     num_groups = (num_row_offsets - 1) // 16 + 1
+    index_bytes = num_groups * ROW_GROUP_SIZE
+    heap_end = len_page - index_bytes
+    if heap_end <= PAGE_HEADER_SIZE:
+        raise UnsupportedFormatError(
+            f"export.pdb page {page_index}: num_row_offsets={num_row_offsets} "
+            f"needs {index_bytes} bytes of row index, which does not fit in a "
+            f"{len_page}-byte page"
+        )
+
     bases: list[int] = []
+    present_count = 0
     for g in range(num_groups):
         group_base = len_page - (g * ROW_GROUP_SIZE)
-        if group_base - 4 < PAGE_HEADER_SIZE:
-            break
         present_flags = struct.unpack_from("<H", page, group_base - 4)[0]
         for r in range(16):
             abs_i = g * 16 + r
             if abs_i >= num_row_offsets:
                 break
             if (present_flags >> r) & 1 == 0:
                 continue  # deleted / absent — honour the bitmask
-            ofs_pos = group_base - (6 + 2 * r)
-            if ofs_pos < 0:
-                continue
-            ofs_row = struct.unpack_from("<H", page, ofs_pos)[0]
+            present_count += 1
+            ofs_row = struct.unpack_from("<H", page, group_base - (6 + 2 * r))[0]
             row_base = ofs_row + PAGE_HEADER_SIZE
-            if 0 <= row_base < len_page:
-                bases.append(row_base)
+            if not (PAGE_HEADER_SIZE <= row_base < heap_end):
+                raise UnsupportedFormatError(
+                    f"export.pdb page {page_index}: row offset {row_base} "
+                    f"(slot {abs_i}) falls outside the row heap "
+                    f"[{PAGE_HEADER_SIZE}, {heap_end}) — the page image is "
+                    f"inconsistent, likely read while being written"
+                )
+            bases.append(row_base)
+
+    if 0 <= num_rows <= _NUM_ROWS_MAX and present_count != num_rows:
+        raise UnsupportedFormatError(
+            f"export.pdb page {page_index}: {present_count} rows marked present "
+            f"but the page header declares num_rows={num_rows} — the page image "
+            f"is inconsistent, likely read while being written"
+        )
     return bases
 
 
@@ -274,14 +317,16 @@ def _walk_table_pages(
 
         next_page = int(lead.next_page)
         page_type = int(lead.page_type)
-        num_row_offsets, _num_rows, page_flags = _decode_page_counts(page)
+        num_row_offsets, num_rows, page_flags = _decode_page_counts(page)
 
         # Stop if we landed on a different table type (safety).
         if page_type != expected_type and idx != first_page:
             break
 
         if _is_data_page(page_flags) and page_type == expected_type:
-            bases = _iter_present_row_bases(page, len_page, num_row_offsets)
+            bases = _iter_present_row_bases(
+                page, len_page, num_row_offsets, num_rows, idx
+            )
             out.append((page, bases))
 
         if idx == last_page:

tests/unit/test_pdb.py

@@ -1109,3 +1109,135 @@ def test_real_export_pdb_scale_and_sanity(tmp_path: Path) -> None:
         f"\nreal_stick: tracks={len(lib.tracks)} playlists={len(lib.playlists)} "
         f"elapsed={elapsed:.3f}s sample={sample.title!r} path={sample.resolved_path}"
     )
+
+
+# ---------------------------------------------------------------------------
+# G1d — torn/mid-export page images
+#
+# A real conversion published two playlist entries that the settled export.pdb
+# does not contain (tracks 444 and 2984, 2026-07-29). The file was unchanged
+# between that convert and the verify that caught it, so the difference arose
+# at read time: convert started 14 s after rekordbox's last write to the pdb.
+# A page whose present-bit set disagrees with its declared num_rows is the
+# signature that resurrects a stale row, and the reader decoded num_rows and
+# then discarded it. These tests pin the gate.
+# ---------------------------------------------------------------------------
+
+
+def _resurrect_slot(
+    page: bytes, len_page: int, slot: int, heap_offset: int
+) -> bytes:
+    """Set slot's present bit and point it at a row body, leaving num_rows alone.
+
+    This is what a torn page image looks like to the walker: one more row
+    reachable than the page header says it holds.
+    """
+    buf = bytearray(page)
+    g, r = divmod(slot, 16)
+    group_base = len_page - (g * 0x24)
+    present = struct.unpack_from("<H", buf, group_base - 4)[0]
+    present |= 1 << r
+    struct.pack_into("<H", buf, group_base - 4, present)
+    struct.pack_into("<H", buf, group_base - (6 + 2 * r), heap_offset)
+    return bytes(buf)
+
+
+def _torn_pdb(tmp_path: Path, entry_page: bytes, len_page: int) -> Path:
+    pages = {
+        0: _file_header(
+            len_page=len_page,
+            tables=[(0, 1, 1), (7, 2, 2), (8, 3, 3)],
+        ),
+        1: _build_nondata_page(
+            len_page=len_page, page_index=1, page_type=0, next_page=99
+        ),
+        2: _build_data_page(
+            len_page=len_page,
+            page_index=2,
+            page_type=7,
+            next_page=99,
+            row_blobs=[
+                _playlist_tree_row(
+                    parent_id=0, sort_order=0, pl_id=1, is_folder=False, name="Set A"
+                )
+            ],
+        ),
+        3: entry_page,
+    }
+    path = tmp_path / "torn.pdb"
+    _write_pdb(path, len_page, pages)
+    return path
+
+
+def test_g1d_resurrected_row_beyond_num_rows_raises(tmp_path: Path) -> None:
+    """A present bit the page header does not account for must be refused.
+
+    WHY: this is the observed failure. The stale entry parses cleanly and lands
+    in a coherent chain, so nothing downstream can tell it from a real one —
+    the conversion exits 0 and publishes a library with a track the DJ removed.
+    Refusing the parse is the only point where the two disagree.
+    """
+    len_page = 512
+    rows: list[bytes | None] = [
+        _playlist_entry_row(0, track_id=10, playlist_id=1),
+        None,  # deleted slot — its bytes are still in the heap
+        _playlist_entry_row(1, track_id=20, playlist_id=1),
+    ]
+    page = _build_data_page(
+        len_page=len_page, page_index=3, page_type=8, next_page=99, row_blobs=rows
+    )
+    # Resurrect slot 1, pointing it at the first row body. num_rows still says 2.
+    torn = _resurrect_slot(page, len_page, slot=1, heap_offset=0)
+    path = _torn_pdb(tmp_path, torn, len_page)
+
+    with pytest.raises(UnsupportedFormatError) as exc:
+        parse_export_pdb(path, tmp_path)
+    assert "num_rows" in str(exc.value) or "present" in str(exc.value)
+
+
+def test_g1d_row_offset_inside_row_index_raises(tmp_path: Path) -> None:
+    """A row offset pointing into the backward-growing index is not a row.
+
+    WHY: across 997 pages of a real 3,673-track export every present row body
+    sits between the page header and the index; nothing legitimate points into
+    the index area. The reader only bounded offsets by the page size, so stale
+    index bytes could be parsed as a row.
+    """
+    len_page = 512
+    rows: list[bytes | None] = [
+        _playlist_entry_row(0, track_id=10, playlist_id=1),
+        _playlist_entry_row(1, track_id=20, playlist_id=1),
+    ]
+    page = _build_data_page(
+        len_page=len_page, page_index=3, page_type=8, next_page=99, row_blobs=rows
+    )
+    buf = bytearray(page)
+    # Point slot 0 into the row-index region (heap-relative → absolute ≥ heap end).
+    struct.pack_into("<H", buf, len_page - 6, len_page - PAGE_HEADER)
+    path = _torn_pdb(tmp_path, bytes(buf), len_page)
+
+    with pytest.raises(UnsupportedFormatError) as exc:
+        parse_export_pdb(path, tmp_path)
+    assert "row offset" in str(exc.value).lower()
+
+
+def test_g1d_well_formed_page_with_deleted_rows_still_parses(tmp_path: Path) -> None:
+    """The gate must not fire on ordinary deleted slots.
+
+    WHY: tombstones are normal — a real export carried 104 of them. A gate that
+    rejected those would refuse every library the tool exists to convert.
+    """
+    len_page = 512
+    rows: list[bytes | None] = [
+        _playlist_entry_row(0, track_id=10, playlist_id=1),
+        None,
+        _playlist_entry_row(1, track_id=20, playlist_id=1),
+    ]
+    page = _build_data_page(
+        len_page=len_page, page_index=3, page_type=8, next_page=99, row_blobs=rows
+    )
+    path = _torn_pdb(tmp_path, page, len_page)
+
+    lib = parse_export_pdb(path, tmp_path)
+
+    assert lib.playlists[0].track_rb_ids == [10, 20]