fix(stats): route KOReader hash collisions safely

Track colliding KOReader statistics rows independently and require verified explicit targets before upload. Keep unresolved history pending with a visible recovery path.

Fixes #1457

koreader-plugin/bookorbit.koplugin/bookorbit_book_sync.lua

@@ -95,7 +95,14 @@ function BookOrbitBookSync.capture(plugin)
     end
     metadata = metadata or {}
     local current_stats_id = stats and tonumber(stats.id_curr_book) or nil
-    local stats_ambiguous = current_stats_id == nil and metadata.metadata_ambiguous == true
+    local stats_ambiguous = current_stats_id == nil and metadata.stats_ambiguous == true
+    local current_stats_row
+    for _, row in ipairs(metadata.rows or {}) do
+        if row.id == current_stats_id then
+            current_stats_row = row
+            break
+        end
+    end
     local stats_ids = {}
     if current_stats_id then
         table.insert(stats_ids, current_stats_id)
@@ -114,6 +121,8 @@ function BookOrbitBookSync.capture(plugin)
         last_open = metadata.last_open or ts,
         metadata_ambiguous = false,
         stats_metadata_ambiguous = stats_ambiguous,
+        stats_row_ambiguous = current_stats_row ~= nil and metadata.stats_ambiguous == true,
+        stats_row = current_stats_row,
         stats_identity_repaired = plugin.bookorbit_document_digest
             and plugin.bookorbit_document_digest.repaired == true or false,
         stats_ids = stats_ids,
@@ -279,6 +288,21 @@ end
 
 local stepMatch, stepStats, stepAnnotations, stepAnnotationsLegacy, stepBookmarks, stepState, stepProgress
 
+local function bindStatsRow(ctx, book)
+    local row = ctx.snap.stats_row
+    if not row or not ctx.snap.stats_row_ambiguous then return nil end
+    return ctx.state:bindStatsRow(
+        row.id, ctx.snap.digest, row.title, row.authors, book.fileId, book.bookId)
+end
+
+local function matchedFileCandidate(ctx, book)
+    if not ctx.snap.stats_row_ambiguous then return book and book.fileId or nil end
+    local row = ctx.snap.stats_row
+    if not row then return nil end
+    local target = ctx.state:getStatsRow(row.id, ctx.snap.digest, row.title, row.authors)
+    return target and target.bookFileId or nil
+end
+
 stepMatch = function(ctx)
     if ctx.acknowledged.match then
         return step(ctx, stepStats)
@@ -288,7 +312,10 @@ stepMatch = function(ctx)
         book.file = ctx.snap.file
     end
 
-    if not ctx.force_match and BookOrbitState.isMatchFresh(book, ctx.state.global) then
+    if not ctx.force_match
+            and not ctx.snap.stats_row_ambiguous
+            and BookOrbitState.isMatchFresh(book, ctx.state.global) then
+        bindStatsRow(ctx, book)
         logger.dbg("BookOrbit: book sync match skipped, verified match still fresh")
         if not acknowledge(ctx, "match", false) then return end
         return step(ctx, stepStats)
@@ -300,7 +327,7 @@ stepMatch = function(ctx)
             authors = ctx.snap.authors,
             last_open = ctx.snap.last_open,
             source = "current_file",
-            book_file_id = book and book.fileId or nil,
+            book_file_id = matchedFileCandidate(ctx, book),
             metadata_ambiguous = ctx.snap.metadata_ambiguous,
         },
     })
@@ -317,6 +344,7 @@ stepMatch = function(ctx)
     for _, match in ipairs(body.matches or {}) do
         if match.hash == ctx.snap.digest then
             ctx.state:setMatched(match.hash, match.bookFileId, match.bookId, ctx.snap.file)
+            bindStatsRow(ctx, ctx.state:getBook(ctx.snap.digest))
             if not acknowledge(ctx, "match") then return end
             return step(ctx, stepStats)
         end
@@ -352,8 +380,9 @@ stepStats = function(ctx)
         return step(ctx, stepAnnotations)
     end
 
+    local stats_target = bindStatsRow(ctx, book) or book
     if ctx.stats_watermark == nil then
-        ctx.stats_watermark = ctx.snap.stats_identity_repaired and 0 or (book.statsWatermark or 0)
+        ctx.stats_watermark = ctx.snap.stats_identity_repaired and 0 or (stats_target.statsWatermark or 0)
     end
     local watermark = ctx.stats_watermark
     local events = BookOrbitStatsReader.getEventsAfter(ctx.stat_ids, watermark, STATS_BATCH)
@@ -362,7 +391,9 @@ stepStats = function(ctx)
         return step(ctx, stepAnnotations)
     end
 
-    local body, err = ctx.client:uploadPageStats({ { hash = ctx.snap.digest, events = events } })
+    local upload = { hash = ctx.snap.digest, events = events }
+    if ctx.snap.stats_row_ambiguous then upload.bookFileId = stats_target.bookFileId end
+    local body, err = ctx.client:uploadPageStats({ upload })
     if not body then
         if isAuthError(err) then return finish(ctx, "auth") end
         ctx.had_errors = true
@@ -375,8 +406,8 @@ stepStats = function(ctx)
         return finish(ctx, "unmatched")
     end
 
-    local more = BookOrbitState.applyStatsAck(book, events, body, ctx.snap.digest, STATS_BATCH, watermark)
-    ctx.stats_watermark = book.statsWatermark or watermark
+    local more = BookOrbitState.applyStatsAck(stats_target, events, body, ctx.snap.digest, STATS_BATCH, watermark)
+    ctx.stats_watermark = stats_target.statsWatermark or watermark
     ctx.counts.page_stats = ctx.counts.page_stats + #events
     if more then
         return step(ctx, stepStats)
@@ -780,6 +811,7 @@ function BookOrbitBookSync.run(opts)
         state = BookOrbitStateManager.session({
             digests = { snap.digest },
             files = { snap.file },
+            statsRows = snap.stats_ids,
         }),
         snap = snap,
         reason = opts.reason or "manual",

koreader-plugin/bookorbit.koplugin/bookorbit_lifecycle_outbox.lua

@@ -406,10 +406,14 @@ function Outbox:enqueue(snapshot, opts)
     end
     local status = self:status()
 
-    local existing
+    local existing, existing_with_payload
     for _, entry in ipairs(self:listMetadata()) do
         if entry.digest == snapshot.digest and entry.id ~= self.active_id then
-            existing = entry
+            local candidate = self:attachPayload(entry)
+            if candidate and candidate.snapshot.file == snapshot.file then
+                existing = entry
+                existing_with_payload = candidate
+            end
         end
     end
     if status.hard_limit and not existing then return nil, "hard_limit" end
@@ -421,7 +425,6 @@ function Outbox:enqueue(snapshot, opts)
     local stored_snapshot = {}
     for key, value in pairs(snapshot) do stored_snapshot[key] = value end
     stored_snapshot.stats_ids = copyArray(snapshot.stats_ids)
-    local existing_with_payload = existing and self:attachPayload(existing) or nil
     if existing_with_payload and existing_with_payload.acknowledged.stats ~= true then
         stored_snapshot.stats_ids = mergeUniqueArrays(
             existing_with_payload.snapshot.stats_ids, stored_snapshot.stats_ids)

koreader-plugin/bookorbit.koplugin/bookorbit_state.lua

@@ -42,6 +42,7 @@ function BookOrbitState.snapshot(tables, flush_handler)
         books = deepCopy(tables.books or {}),
         unmatched = deepCopy(tables.unmatched or {}),
         files = deepCopy(tables.files or {}),
+        statsRows = deepCopy(tables.statsRows or {}),
         global = deepCopy(tables.global or {}),
         flush_handler = flush_handler,
     }, BookOrbitState)
@@ -57,6 +58,7 @@ function BookOrbitState:reload()
     self.books = settings:readSetting("books", {})
     self.unmatched = settings:readSetting("unmatched", {})
     self.files = settings:readSetting("files", {})
+    self.statsRows = settings:readSetting("statsRows", {})
     self.global = settings:readSetting("global", {})
     return self
 end
@@ -65,6 +67,40 @@ function BookOrbitState:getBook(md5)
     return self.books[md5]
 end
 
+-- A partial MD5 may identify several KOReader statistics rows. Those rows
+-- need independent cursors: advancing one book's watermark must never hide
+-- an older event belonging to another book with the same digest. Row ids are
+-- local to statistics.sqlite3, so the persisted metadata guards against an id
+-- being reused after that database is rebuilt.
+function BookOrbitState:getStatsRow(id, md5, title, authors)
+    id = tonumber(id)
+    if not id or not md5 then return nil end
+    local key = tostring(id)
+    local row = self.statsRows[key]
+    if not row or row.md5 ~= md5 or row.title ~= title or row.authors ~= authors then
+        row = {
+            md5 = md5,
+            title = title,
+            authors = authors,
+            statsWatermark = 0,
+        }
+        self.statsRows[key] = row
+    end
+    row.statsWatermark = row.statsWatermark or 0
+    return row
+end
+
+function BookOrbitState:bindStatsRow(id, md5, title, authors, book_file_id, book_id)
+    local row = self:getStatsRow(id, md5, title, authors)
+    if not row then return nil end
+    if row.bookFileId ~= book_file_id or row.bookId ~= book_id then
+        row.statsWatermark = 0
+    end
+    row.bookFileId = book_file_id
+    row.bookId = book_id
+    return row
+end
+
 -- Maps BookOrbit bookId -> local file path and bookFileId -> local file path
 -- for every matched book whose file is still present on disk. Both maps come
 -- from one scan with one existence check per stored file; bookorbit_state_manager

koreader-plugin/bookorbit.koplugin/bookorbit_state_manager.lua

@@ -123,11 +123,12 @@ local function scopeTables(state, scope)
             books = state.books,
             unmatched = state.unmatched,
             files = state.files,
+            statsRows = state.statsRows,
             global = state.global,
         }
     end
 
-    local tables = { books = {}, unmatched = {}, files = {}, global = {} }
+    local tables = { books = {}, unmatched = {}, files = {}, statsRows = {}, global = {} }
     for digest in pairs(scope.digests) do
         tables.books[digest] = state.books[digest]
         tables.unmatched[digest] = state.unmatched[digest]
@@ -135,21 +136,28 @@ local function scopeTables(state, scope)
     for file in pairs(scope.files) do
         tables.files[file] = state.files[file]
     end
+    for row_id in pairs(scope.statsRows) do
+        tables.statsRows[row_id] = state.statsRows[row_id]
+    end
     if scope.global then tables.global = state.global end
     return tables
 end
 
 local function normalizeScope(opts)
     if not opts or opts.full then
-        return { full = true, digests = {}, files = {}, global = true }
+        return { full = true, digests = {}, files = {}, statsRows = {}, global = true }
     end
-    local scope = { full = false, digests = {}, files = {}, global = opts.global ~= false }
+    local scope = { full = false, digests = {}, files = {}, statsRows = {}, global = opts.global ~= false }
     for _, digest in ipairs(opts.digests or {}) do
         if digest then scope.digests[digest] = true end
     end
     for _, file in ipairs(opts.files or {}) do
         if file then scope.files[file] = true end
     end
+    for _, row_id in ipairs(opts.statsRows or {}) do
+        row_id = tonumber(row_id)
+        if row_id then scope.statsRows[tostring(row_id)] = true end
+    end
     return scope
 end
 
@@ -158,6 +166,7 @@ local function refreshSession(session)
     session.books = deepCopy(current.books)
     session.unmatched = deepCopy(current.unmatched)
     session.files = deepCopy(current.files)
+    session.statsRows = deepCopy(current.statsRows)
     session.global = deepCopy(current.global)
     session.manager_base = deepCopy(current)
     session.manager_generation = generation
@@ -172,6 +181,7 @@ local function commitSession(session)
         replaceTable(state.books, mergeValue(base.books, session.books, state.books))
         replaceTable(state.unmatched, mergeValue(base.unmatched, session.unmatched, state.unmatched))
         replaceTable(state.files, mergeValue(base.files, session.files, state.files))
+        replaceTable(state.statsRows, mergeValue(base.statsRows, session.statsRows, state.statsRows))
         replaceTable(state.global, mergeValue(base.global, session.global, state.global))
     else
         for digest in pairs(scope.digests) do
@@ -182,6 +192,9 @@ local function commitSession(session)
         for file in pairs(scope.files) do
             state.files[file] = mergeValue(base.files[file], session.files[file], state.files[file])
         end
+        for row_id in pairs(scope.statsRows) do
+            state.statsRows[row_id] = mergeValue(base.statsRows[row_id], session.statsRows[row_id], state.statsRows[row_id])
+        end
         if scope.global then
             replaceTable(state.global, mergeValue(base.global, session.global, state.global))
         end

koreader-plugin/bookorbit.koplugin/bookorbit_stats_reader.lua

@@ -38,6 +38,12 @@ local function applyRow(entry, id, title, authors, last_open)
     if title == "" then title = nil end
     if authors == "" then authors = nil end
     table.insert(entry.ids, id)
+    table.insert(entry.rows, {
+        id = id,
+        title = title,
+        authors = authors,
+        last_open = last_open,
+    })
     if not entry._variant_seen[metadataKey(title, authors)] then
         entry._variant_seen[metadataKey(title, authors)] = true
         entry._variant_count = entry._variant_count + 1
@@ -53,6 +59,7 @@ end
 
 local function finalizeEntry(entry)
     entry.metadata_ambiguous = (entry._variant_count or 0) > 1
+    entry.stats_ambiguous = #entry.ids > 1
     entry._variant_seen = nil
     entry._variant_count = nil
     return entry
@@ -163,7 +170,7 @@ function Session:book(md5)
     if type(md5) ~= "string" or not md5:match("^%x+$") then return nil end
     local res = self:select(BOOK_BY_MD5_SQL, { md5 })
     if not res then return nil end
-    local book = { ids = {}, last_open = 0, _variant_seen = {}, _variant_count = 0 }
+    local book = { ids = {}, rows = {}, last_open = 0, _variant_seen = {}, _variant_count = 0 }
     for i = 1, #res[1] do
         applyRow(book, tonumber(res[1][i]), res[2][i], res[3][i], tonumber(res[4][i]) or 0)
     end
@@ -318,7 +325,7 @@ end
 function BookOrbitStatsReader.collectBookRow(collector, row)
     local entry = collector.by_md5[row.md5]
     if not entry then
-        entry = { md5 = row.md5, ids = {}, last_open = 0, _variant_seen = {}, _variant_count = 0 }
+        entry = { md5 = row.md5, ids = {}, rows = {}, last_open = 0, _variant_seen = {}, _variant_count = 0 }
         collector.by_md5[row.md5] = entry
         table.insert(collector.list, entry)
     end
@@ -334,7 +341,7 @@ function BookOrbitStatsReader.finalizeBookEntry(entry)
     return finalizeEntry(entry)
 end
 
--- Returns { ids, last_open, title, authors, metadata_ambiguous } for one md5.
+-- Returns { ids, rows, last_open, title, authors, metadata_ambiguous, stats_ambiguous } for one md5.
 -- Used by live single-book sync before match-check.
 function BookOrbitStatsReader.getBook(md5)
     return withSession(function(session)

koreader-plugin/bookorbit.koplugin/bookorbit_sweep.lua

@@ -163,9 +163,10 @@ end
 
 local function chunkSummary(ctx)
     return string.format(
-        "enumerateMs=%.0f partialMd5Ms=%.0f partialMd5Count=%d sidecarMs=%.0f slowestItemMs=%.0f statsSkipped=%d",
+        "enumerateMs=%.0f partialMd5Ms=%.0f partialMd5Count=%d sidecarMs=%.0f slowestItemMs=%.0f statsSkipped=%d ambiguousStats=%d",
         ctx.timing.enumerate_ms, ctx.timing.partial_md5_ms, ctx.timing.partial_md5_count,
-        ctx.timing.sidecar_ms, ctx.timing.max_item_ms, ctx.counts.stats_skipped or 0)
+        ctx.timing.sidecar_ms, ctx.timing.max_item_ms, ctx.counts.stats_skipped or 0,
+        ctx.counts.stats_ambiguous or 0)
 end
 
 local function finish(ctx, err)
@@ -204,10 +205,15 @@ local function finish(ctx, err)
         elseif not err and ctx.had_errors then
             ctx.plugin:recordSyncError("sweep", "partial_failure")
         elseif not err and ctx.plugin.recordSyncSuccess then
+            local message = T(_("%1 books, %2 reading events, %3 highlights"),
+                ctx.counts.books_matched, ctx.counts.page_stats, ctx.counts.annotations)
+            if (ctx.counts.stats_ambiguous or 0) > 0 then
+                message = message .. ". " .. T(_("%1 books need to be opened and synced once to recover their reading history."),
+                    ctx.counts.stats_ambiguous)
+            end
             ctx.plugin:recordSyncSuccess(
                 "sweep",
-                T(_("%1 books, %2 reading events, %3 highlights"),
-                    ctx.counts.books_matched, ctx.counts.page_stats, ctx.counts.annotations))
+                message)
         end
     end
 
@@ -270,6 +276,10 @@ local function finish(ctx, err)
         if ctx.had_errors then
             text = text .. "\n" .. _("Some books failed and will retry on the next sync.")
         end
+        if (ctx.counts.stats_ambiguous or 0) > 0 then
+            text = text .. "\n" .. T(_("%1 books need to be opened and synced once to recover their reading history."),
+                ctx.counts.stats_ambiguous)
+        end
         UIManager:show(InfoMessage:new{ text = text, timeout = 5 })
     end
     logger.info(string.format(
@@ -421,12 +431,13 @@ local function stepEnumerateCandidates(ctx)
         BookOrbitStatsReader.finalizeBookEntry(entry)
         chunk_ctx.candidates[entry.md5] = {
             stat_ids = entry.ids,
+            stats_rows = entry.rows,
             last_open = entry.last_open,
             title = entry.title,
             authors = entry.authors,
             source = "statistics",
             metadata_ambiguous = entry.metadata_ambiguous,
-            stats_metadata_ambiguous = entry.metadata_ambiguous,
+            stats_metadata_ambiguous = entry.stats_ambiguous,
         }
     end
 
@@ -600,11 +611,30 @@ local function buildStatsQueue(ctx)
     local latest = ctx.stats_session and ctx.stats_session:latestEventTimes() or nil
     ctx.stats_queue = BookOrbitQueue.new()
     ctx.counts.stats_skipped = 0
+    ctx.counts.stats_ambiguous = 0
 
     for md5, cand in pairs(ctx.candidates) do
-        if cand.stat_ids and not cand.stats_metadata_ambiguous then
-            local book = ctx.state:getBook(md5)
-            if book then
+        local book = ctx.state:getBook(md5)
+        if cand.stat_ids and book then
+            if cand.stats_metadata_ambiguous then
+                for _, row in ipairs(cand.stats_rows or {}) do
+                    local target = ctx.state:getStatsRow(row.id, md5, row.title, row.authors)
+                    if target.bookFileId then
+                        if latest and not hasNewEvents(latest, { row.id }, target.statsWatermark or 0) then
+                            ctx.counts.stats_skipped = ctx.counts.stats_skipped + 1
+                        else
+                            ctx.stats_queue:push({
+                                md5 = md5,
+                                ids = { row.id },
+                                target = target,
+                                book_file_id = target.bookFileId,
+                            })
+                        end
+                    elseif not latest or (latest[row.id] or 0) > (target.statsWatermark or 0) then
+                        ctx.counts.stats_ambiguous = ctx.counts.stats_ambiguous + 1
+                    end
+                end
+            else
                 if latest and not hasNewEvents(latest, cand.stat_ids, book.statsWatermark or 0) then
                     ctx.counts.stats_skipped = ctx.counts.stats_skipped + 1
                 else
@@ -681,7 +711,8 @@ local function stepStatsNext(ctx)
         return
     end
 
-    local watermark = book.statsWatermark or 0
+    local stats_target = item.target or book
+    local watermark = stats_target.statsWatermark or 0
     local events = ctx.stats_session
         and ctx.stats_session:eventsAfter(item.ids, watermark, STATS_BATCH)
         or BookOrbitStatsReader.getEventsAfter(item.ids, watermark, STATS_BATCH)
@@ -690,7 +721,9 @@ local function stepStatsNext(ctx)
         return
     end
 
-    local body, err = ctx.client:uploadPageStats({ { hash = item.md5, events = events } })
+    local upload = { hash = item.md5, events = events }
+    if item.book_file_id then upload.bookFileId = item.book_file_id end
+    local body, err = ctx.client:uploadPageStats({ upload })
     if aborted(ctx) then return end
     if not body then
         if isAuthError(err) then return finish(ctx, "auth") end
@@ -702,13 +735,20 @@ local function stepStatsNext(ctx)
 
     for _, unmatched in ipairs(body.unmatched or {}) do
         if unmatched == item.md5 then
-            ctx.state:setUnmatched(item.md5)
+            if item.book_file_id then
+                stats_target.bookFileId = nil
+                stats_target.bookId = nil
+                stats_target.statsWatermark = 0
+                ctx.had_errors = true
+            else
+                ctx.state:setUnmatched(item.md5)
+            end
             step(ctx, ctx.steps.statsNext)
             return
         end
     end
 
-    local more = BookOrbitState.applyStatsAck(book, events, body, item.md5, STATS_BATCH, watermark)
+    local more = BookOrbitState.applyStatsAck(stats_target, events, body, item.md5, STATS_BATCH, watermark)
     ctx.counts.page_stats = ctx.counts.page_stats + #events
     if more then
         ctx.current_stats = item

koreader-plugin/bookorbit.koplugin/main.lua

@@ -48,7 +48,7 @@ local BookOrbitProgressSync = require("bookorbit_progress_sync")
 local BookOrbitSweep = require("bookorbit_sweep")
 local BookOrbitUpdater = require("bookorbit_updater")
 
-local PLUGIN_VERSION = "1.5.3"
+local PLUGIN_VERSION = "1.5.4"
 
 local SYNC_STRATEGY = {
     PROMPT = 1,

koreader-plugin/spec/bookorbit_book_sync_fast_path_test.lua

@@ -11,6 +11,9 @@ local calls
 local last_match_candidates
 local stats_watermarks
 local page_stat_events
+local last_stats_upload
+local bound_stats_row
+local stored_stats_row
 local match_fresh = false
 local exchange_skippable = false
 
@@ -55,9 +58,10 @@ package.loaded["bookorbit_api"] = {
             matchCheck = function(_, _, candidates)
                 table.insert(calls, "match")
                 last_match_candidates = candidates
+                local target = candidates.abcdef.book_file_id or 2
                 return {
                     libraryVersion = "lib-v1",
-                    matches = { { hash = "abcdef", bookFileId = 2, bookId = 1 } },
+                    matches = { { hash = "abcdef", bookFileId = target, bookId = target + 10 } },
                 }
             end,
             uploadBookStates = function()
@@ -66,6 +70,7 @@ package.loaded["bookorbit_api"] = {
             end,
             uploadPageStats = function(_, books)
                 table.insert(calls, "stats")
+                last_stats_upload = books[1]
                 return {
                     results = { { hash = books[1].hash, watermark = books[1].events[#books[1].events].startTime } },
                     unmatched = {},
@@ -103,9 +108,29 @@ local book
 local state = {
     global = { libraryVersion = "lib-v1" },
     getBook = function() return book end,
-    setMatched = function() end,
+    setMatched = function(_, _, book_file_id, book_id, file)
+        book.fileId = book_file_id
+        book.bookId = book_id
+        book.file = file or book.file
+    end,
     setUnmatched = function() book = nil end,
     rememberFile = function() end,
+    getStatsRow = function(_, id, md5, title, authors)
+        if not stored_stats_row then
+            stored_stats_row = {
+                id = id, md5 = md5, title = title, authors = authors, statsWatermark = 0,
+            }
+        end
+        return stored_stats_row
+    end,
+    bindStatsRow = function(_, id, md5, title, authors, book_file_id, book_id)
+        bound_stats_row = {
+            id = id, md5 = md5, title = title, authors = authors,
+            bookFileId = book_file_id, bookId = book_id, statsWatermark = 0,
+        }
+        stored_stats_row = bound_stats_row
+        return bound_stats_row
+    end,
     flush = function() end,
 }
 package.loaded["bookorbit_state_manager"] = { session = function() return state end }
@@ -137,6 +162,9 @@ local function run(opts)
     calls = {}
     stats_watermarks = {}
     page_stat_events = opts.page_stat_events or {}
+    last_stats_upload = nil
+    bound_stats_row = nil
+    stored_stats_row = opts.stored_stats_row
     book = {
         bookId = 1,
         fileId = 2,
@@ -153,6 +181,8 @@ local function run(opts)
             file = "/books/a.epub",
             stats_ids = { 42 },
             stats_identity_repaired = opts.stats_identity_repaired == true,
+            stats_row_ambiguous = opts.stats_row_ambiguous == true,
+            stats_row = opts.stats_row,
             annotations = {},
             ann_count = 0,
             ann_signature = "0::0:0",
@@ -223,4 +253,36 @@ requests = run{
 assertEqual(stats_watermarks[1], 0, "identity repair replays exact-row history before the stored watermark")
 assertEqual(requests, "stats,progress", "the stranded historical event is uploaded under the repaired digest")
 
+match_fresh = true
+requests = run{
+    stats_row_ambiguous = true,
+    stats_row = { id = 42, title = "Collision Two", authors = "Author B" },
+    stats_watermark = 900,
+    page_stat_events = { { page = 2, startTime = 600, durationSeconds = 60, totalPages = 100 } },
+}
+assertEqual(bound_stats_row.id, 42, "the exact live KOReader statistics row is bound")
+assertEqual(bound_stats_row.bookFileId, 2, "the row is bound to the verified server file")
+assertEqual(stats_watermarks[1], 0, "a newly bound collision row replays its own complete history")
+assertEqual(last_stats_upload.bookFileId, 2, "collision events carry the explicit verified server target")
+assertEqual(last_match_candidates.abcdef.book_file_id, nil,
+    "a collision match never proposes the sibling hash-level file target")
+assertEqual(requests, "match,stats,progress",
+    "a collision row revalidates its target instead of trusting the shared hash cache")
+
+requests = run{
+    stats_row_ambiguous = true,
+    stats_row = { id = 42, title = "Collision Two", authors = "Author B" },
+    stored_stats_row = {
+        id = 42,
+        md5 = "abcdef",
+        title = "Collision Two",
+        authors = "Author B",
+        bookFileId = 7,
+        bookId = 17,
+        statsWatermark = 500,
+    },
+}
+assertEqual(last_match_candidates.abcdef.book_file_id, 7,
+    "an established row-specific target is reused for collision revalidation")
+
 print("bookorbit_book_sync_fast_path_test.lua: ok")

koreader-plugin/spec/bookorbit_capture_identity_test.lua

@@ -123,6 +123,8 @@ assertEqual(sqlite.opens, 1, "capture opens no database when identity is primed"
 assertEqual(flushes, 1, "the pending page-stat flush still runs live at capture")
 assertEqual(#snap.stats_ids, 1, "the snapshot carries the statistics row ids")
 assertEqual(snap.stats_ids[1], 42, "the row id comes from the primed identity")
+assertEqual(snap.stats_row.id, 42, "the exact live statistics row is carried with its metadata")
+assertEqual(snap.stats_row_ambiguous, false, "a single statistics row needs no explicit collision routing")
 assertEqual(snap.title, "Statistics Title", "stable identity fields come from the cache")
 assertEqual(snap.percentage, 0.5, "progress is read live, never cached")
 assert(type(snap.ann_signature) == "string" and snap.ann_signature ~= "",
@@ -145,6 +147,22 @@ local warm = BookSync.capture(plugin)
 assertEqual(sqlite.opens, 1, "the fallback primes the cache for the next handler")
 assertEqual(warm.stats_ids[1], 42, "the warmed capture carries the same row ids")
 
+-- A genuine partial-MD5 collision still has an exact live KOReader row while
+-- the book is open. Capture must preserve that row rather than the whole hash
+-- group so the drain can bind and replay it safely.
+book_rows = {
+    { "42", "Collision One", "Author A", "1700000000" },
+    { "43", "Collision Two", "Author B", "1700000100" },
+}
+plugin.ui.statistics.id_curr_book = 43
+BookOrbitStatsReader.forgetIdentity()
+local collision = BookSync.capture(plugin)
+assertEqual(collision.stats_row_ambiguous, true, "the exact row records that its digest collides")
+assertEqual(collision.stats_row.id, 43, "capture selects the live collision row")
+assertEqual(collision.stats_row.title, "Collision Two", "the selected row keeps its identity guard")
+assertEqual(#collision.stats_ids, 1, "only the live collision row is eligible for this book sync")
+assertEqual(collision.stats_ids[1], 43, "the other book's statistics row is never included")
+
 -- A repaired digest can be a cached miss while KOReader still has historical
 -- events attached to the current row's old digest. The live row id is exact,
 -- so capture carries it under the recomputed file hash without a second query.

koreader-plugin/spec/bookorbit_lifecycle_outbox_test.lua

@@ -65,6 +65,15 @@ assertEqual(replacement.snapshot.stats_ids[3], 29, "the newest statistics row is
 assertEqual(replacement.snapshot.stats_identity_repaired, true,
     "a pending identity-repair replay survives snapshot replacement")
 
+local collision_snapshot = {}
+for key, value in pairs(snapshot) do collision_snapshot[key] = value end
+collision_snapshot.file = "/books/collision.epub"
+collision_snapshot.stats_ids = { 41 }
+local collision = assert(outbox:enqueue(collision_snapshot, { reason = "close", annotation_sync = true }))
+assert(collision.id ~= replacement.id, "colliding files keep independent outbox entries")
+assertEqual(outbox:status().count, 2, "a shared digest cannot coalesce distinct files")
+outbox:removeEntry(collision.id)
+
 fixture.fail_write = true
 local failed, failure = outbox:enqueue(snapshot, { reason = "close" })
 fixture.fail_write = false

koreader-plugin/spec/bookorbit_match_freshness_test.lua

@@ -164,6 +164,26 @@ assert(unscoped:getBook("legacy-cursor").annWatermark == "",
 replaced:setUnmatched("moved")
 assert(replaced:getBook("moved") == nil, "an unmatched response removes the complete matched record")
 
+-- Partial-MD5 collisions use one guarded cursor per KOReader statistics row.
+-- Rebinding a row to a different server file replays it, while another row
+-- sharing the same digest remains independent.
+local collision = newState({ libraryVersion = "lib-v1" })
+local first_row = collision:bindStatsRow(41, "shared", "First", "Author A", 11, 1)
+first_row.statsWatermark = 500
+local second_row = collision:bindStatsRow(42, "shared", "Second", "Author B", 22, 2)
+second_row.statsWatermark = 600
+assert(collision:getStatsRow(41, "shared", "First", "Author A").statsWatermark == 500,
+    "the first collision row keeps its own watermark")
+assert(collision:getStatsRow(42, "shared", "Second", "Author B").statsWatermark == 600,
+    "the second collision row keeps its own watermark")
+collision:bindStatsRow(41, "shared", "First", "Author A", 33, 3)
+assert(collision:getStatsRow(41, "shared", "First", "Author A").statsWatermark == 0,
+    "a changed explicit target replays that row")
+assert(collision:getStatsRow(42, "shared", "Second", "Author B").statsWatermark == 600,
+    "rebinding one collision row does not reset another")
+assert(collision:getStatsRow(42, "shared", "Second renamed", "Author B").bookFileId == nil,
+    "changed row metadata invalidates a potentially reused local row id")
+
 -- State written before the stamp existed must recheck once, not be trusted.
 local legacy = newState({ libraryVersion = "lib-v1" })
 legacy.books["legacy1"] = { bookId = 3, fileId = 4, file = "/books/legacy.epub" }

koreader-plugin/spec/bookorbit_stats_reader_test.lua

@@ -45,6 +45,46 @@ do
     assertEqual(stats.queries, 0, "and never reaches the database")
 end
 
+-- Collision groups retain each row's metadata, not just the combined id list,
+-- so callers can persist a guarded per-row server identity and cursor.
+do
+    local Reader = load(function(sql)
+        if sql:find("FROM book WHERE md5", 1, true) then
+            return FakeSqlite.resultSet{
+                { "7", "Collision One", "Author A", "100" },
+                { "9", "Collision Two", "Author B", "200" },
+            }
+        end
+        return nil
+    end)
+
+    local book = Reader.getBook("abc123")
+    assertEqual(book.metadata_ambiguous, true, "different metadata marks the digest ambiguous")
+    assertEqual(book.stats_ambiguous, true, "multiple rows mark statistics routing ambiguous")
+    assertEqual(#book.rows, 2, "both guarded row identities are retained")
+    assertEqual(book.rows[1].id, 7, "the first row id is numeric")
+    assertEqual(book.rows[2].title, "Collision Two", "row metadata stays attached to its id")
+    assertEqual(book.rows[2].last_open, 200, "row activity remains available independently")
+end
+
+-- SQLite UNIQUE constraints permit repeated NULL values, so the collision
+-- guard must count rows rather than relying on distinct metadata variants.
+do
+    local Reader = load(function(sql)
+        if sql:find("FROM book WHERE md5", 1, true) then
+            return FakeSqlite.resultSet{
+                { "7", "Same Title", nil, "100" },
+                { "9", "Same Title", nil, "200" },
+            }
+        end
+        return nil
+    end)
+
+    local book = Reader.getBook("abc123")
+    assertEqual(book.metadata_ambiguous, false, "identical metadata remains usable for matching")
+    assertEqual(book.stats_ambiguous, true, "two rows still require independent statistics routing")
+end
+
 -- Event batching binds the row ids, then the watermark, then the limit.
 do
     local seen

koreader-plugin/spec/bookorbit_sweep_test.lua

@@ -288,6 +288,101 @@ do
         "sweep recovery carries only the actual digest's verified file id")
 end
 
+-- Statistics rows that share a partial MD5 keep independent server targets
+-- and watermarks. The older cursor for one book must not hide newer reading
+-- from another book in the collision group.
+do
+    local hash = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+    local harness = SweepHarness.install{
+        books = {
+            { md5 = hash, id = 41, title = "Collision One", authors = "Author A", last_open = 500 },
+            { md5 = hash, id = 42, title = "Collision Two", authors = "Author B", last_open = 600 },
+        },
+        events = {
+            [41] = { { page = 1, start_time = 500 } },
+            [42] = { { page = 2, start_time = 600 } },
+        },
+        library_version = "v1",
+        state = {
+            books = {
+                [hash] = {
+                    bookId = 1, fileId = 11, statsWatermark = 900,
+                    matchVerifiedAt = NOW, matchVerifiedVersion = "v1",
+                },
+            },
+            statsRows = {
+                ["41"] = {
+                    md5 = hash, title = "Collision One", authors = "Author A",
+                    bookId = 1, bookFileId = 11, statsWatermark = 550,
+                },
+                ["42"] = {
+                    md5 = hash, title = "Collision Two", authors = "Author B",
+                    bookId = 2, bookFileId = 22, statsWatermark = 0,
+                },
+            },
+            global = { libraryVersion = "v1" },
+        },
+        page_stats_response = function(books)
+            return {
+                results = { { hash = books[1].hash, watermark = books[1].events[1].startTime } },
+                unmatched = {},
+            }
+        end,
+    }
+    local Sweep = require("bookorbit_sweep")
+
+    startSweep(harness, Sweep)
+    harness.scheduler:drain()
+
+    assertEqual(#harness.calls.page_stats, 1, "only the collision row with new events uploads")
+    assertEqual(harness.calls.page_stats[1][1].bookFileId, 22,
+        "the collision row carries its independently verified server file id")
+    assertEqual(harness.calls.page_stats[1][1].events[1].startTime, 600,
+        "the other collision row's newer watermark does not suppress this event")
+    assertEqual(harness.state.statsRows["41"].statsWatermark, 550,
+        "the unchanged collision row keeps its own watermark")
+    assertEqual(harness.state.statsRows["42"].statsWatermark, 600,
+        "the uploaded collision row advances only its own watermark")
+end
+
+-- Never guess a server target for a collision row. It stays pending and the
+-- successful sweep tells the user exactly how to establish the missing link.
+do
+    local hash = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+    local harness = SweepHarness.install{
+        books = {
+            { md5 = hash, id = 51, title = "Known", authors = "Author A", last_open = 500 },
+            { md5 = hash, id = 52, title = "Needs Link", authors = "Author B", last_open = 600 },
+        },
+        events = { [52] = { { page = 2, start_time = 600 } } },
+        library_version = "v1",
+        state = {
+            books = {
+                [hash] = {
+                    bookId = 1, fileId = 11, statsWatermark = 0,
+                    matchVerifiedAt = NOW, matchVerifiedVersion = "v1",
+                },
+            },
+            statsRows = {
+                ["51"] = {
+                    md5 = hash, title = "Known", authors = "Author A",
+                    bookId = 1, bookFileId = 11, statsWatermark = 0,
+                },
+            },
+            global = { libraryVersion = "v1" },
+        },
+    }
+    local Sweep = require("bookorbit_sweep")
+
+    startSweep(harness, Sweep)
+    harness.scheduler:drain()
+
+    assertEqual(#harness.calls.page_stats, 0, "an unbound collision row is never attributed by guesswork")
+    assert(harness.shown.text:find("1 books need to be opened and synced once", 1, true),
+        "the completed sweep exposes the recovery action")
+    assertEqual(harness.calls.sweep_complete, 1, "the diagnostic does not turn safe deferral into a failed sweep")
+end
+
 -- Cancellation stops the run at its next yield: the step already scheduled
 -- runs, sees it is no longer the current generation and writes nothing.
 do

koreader-plugin/spec/helpers/sweep_harness.lua

@@ -127,7 +127,7 @@ opts:
 - books: statistics rows, { md5, id, title, authors, last_open }
 - events: map of statistics row id to { page, start_time, duration, total_pages }
 - history: ReadHistory entries, { file, time, text }
-- state: initial { books, unmatched, files, global }
+-- state: initial { books, unmatched, files, statsRows, global }
 - library_version: token every server response carries
 - unmatched: set of hashes match-check refuses to match
 - matches: map of hashes to custom { bookId, bookFileId } match results
@@ -299,6 +299,7 @@ function SweepHarness.install(opts)
         books = initial.books or {},
         unmatched = initial.unmatched or {},
         files = initial.files or {},
+        statsRows = initial.statsRows or {},
         global = initial.global or {},
     }, function()
         handle.flushes = handle.flushes + 1