fix(koplugin): key library pull cursor on synced_at to stop stale library (#4934) (#4944)

The KOReader plugin's incremental books pull went permanently stale: after
a while it stopped receiving any updates made from other devices, and only
deleting readest_library.sqlite3 + "Pull books now" recovered it (until it
re-broke). The iOS/web library was unaffected.

Root cause: since #4678 the server keys the books pull on the server-stamped
synced_at column, and the web client (computeMaxTimestamp) advances its
cursor from synced_at. The koplugin was left on updated_at: pullBooks
advanced last_books_pulled_at from max(updated_at, deleted_at). updated_at
is client-supplied, and the koplugin bumps it from the device clock
(touchBook = os.time()*1000). An e-reader clock ahead of the server (or any
row anywhere carrying a future updated_at) drove the cursor past server-now,
so the server's synced_at > since filter returned nothing forever.

The cursor was also shared between the pull side (compared vs server
synced_at) and push-delta detection (getChangedBooks vs local updated_at),
so it could not simply be retargeted.

Fix:
- parseSyncRow reads synced_at; new row_pull_cursor() prefers it and falls
  back to max(updated_at, deleted_at) for a pre-synced_at server, mirroring
  computeMaxTimestamp.
- Split the cursor: last_books_pulled_at now tracks server synced_at (pull
  only); new last_books_pushed_at tracks local updated_at for getChangedBooks
  and is advanced on both pull and push, preserving push dedup.
- v2 -> v3 migration seeds the push watermark from the old shared value and
  resets the pull cursor to 0, auto-healing already-stale installs with one
  full re-pull (no manual sqlite deletion) and no re-push storm.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

apps/readest.koplugin/library/librarystore.lua

@@ -13,7 +13,7 @@
 local SQ3 = require("lua-ljsqlite3/init")
 local json = require("json")
 
-local SCHEMA_VERSION = 2
+local SCHEMA_VERSION = 3
 
 local SCHEMA_SQL = [[
 CREATE TABLE IF NOT EXISTS books (
@@ -152,6 +152,23 @@ function M.new(opts)
             self.db:exec("ALTER TABLE books ADD COLUMN reading_status_updated_at INTEGER;")
         end)
     end
+    -- v2 -> v3: split the shared books cursor into a pull cursor (server
+    -- synced_at) and a push watermark (local updated_at). The old shared
+    -- `last_books_pulled_at` was advanced from client updated_at, which a
+    -- device with a fast clock could push into the future, starving every
+    -- later synced_at-keyed pull so the library went stale and never
+    -- recovered (issue #4934). Seed the new push watermark from the old value
+    -- (local-change detection is unbroken), then zero the pull cursor so the
+    -- next sync re-establishes it on the server's synced_at clock.
+    if prev >= 1 and prev < 3 then
+        self.db:exec([[
+            INSERT INTO sync_state (user_id, key, value)
+                SELECT user_id, 'last_books_pushed_at', value
+                  FROM sync_state WHERE key = 'last_books_pulled_at'
+            ON CONFLICT(user_id, key) DO UPDATE SET value = excluded.value;
+            UPDATE sync_state SET value = '0' WHERE key = 'last_books_pulled_at';
+        ]])
+    end
     if prev < SCHEMA_VERSION then
         self.db:exec(string.format("PRAGMA user_version = %d;", SCHEMA_VERSION))
     end
@@ -189,6 +206,30 @@ function M:setLastPulledAt(ts)
     stmt:close()
 end
 
+-- The push watermark: max local updated_at | deleted_at we've already
+-- accounted for (pushed to the server or pulled from it). getChangedBooks
+-- keys on this so a book already on the server isn't re-pushed. It is kept
+-- SEPARATE from the pull cursor (last_books_pulled_at, server synced_at)
+-- because the two live on different clocks — mixing them starved pulls in
+-- issue #4934.
+function M:getLastPushedAt()
+    local stmt = self.db:prepare(
+        "SELECT value FROM sync_state WHERE user_id = ? AND key = ?")
+    local row = stmt:reset():bind(self.user_id, "last_books_pushed_at"):step()
+    stmt:close()
+    if not row then return nil end
+    return tonumber(row[1])
+end
+
+function M:setLastPushedAt(ts)
+    local stmt = self.db:prepare([[
+        INSERT INTO sync_state (user_id, key, value) VALUES (?, ?, ?)
+        ON CONFLICT(user_id, key) DO UPDATE SET value = excluded.value
+    ]])
+    stmt:reset():bind(self.user_id, "last_books_pushed_at", tostring(ts)):step()
+    stmt:close()
+end
+
 -- ---------------------------------------------------------------------------
 -- upsertBook
 -- ---------------------------------------------------------------------------
@@ -765,6 +806,9 @@ function M.parseSyncRow(dbRow)
         updated_at   = iso_to_ms(dbRow.updated_at),
         created_at   = iso_to_ms(dbRow.created_at),
         deleted_at   = iso_to_ms(dbRow.deleted_at),
+        -- Server-stamped pull cursor (issue #4678); transient — used only to
+        -- advance last_books_pulled_at, not persisted as a books column.
+        synced_at    = iso_to_ms(dbRow.synced_at),
     }
 
     -- Metadata: parse JSON string OR accept an already-parsed table; extract

apps/readest.koplugin/library/syncbooks.lua

@@ -172,6 +172,25 @@ local function row_to_wire(row)
 end
 M._row_to_wire = row_to_wire  -- exported for tests
 
+-- ---------------------------------------------------------------------------
+-- row_pull_cursor(parsed) — this parsed row's contribution to the
+-- incremental-pull watermark. The server keys the books pull on the
+-- server-stamped `synced_at` (issue #4678), so when present it wins outright:
+-- updated_at/deleted_at are client-supplied and can run ahead of the server
+-- on a device with a fast clock, which pushed the cursor into the future and
+-- starved every later pull (issue #4934). Rows from a pre-synced_at server
+-- fall back to max(updated_at, deleted_at). Mirrors computeMaxTimestamp at
+-- apps/readest-app/src/hooks/useSync.ts:29.
+-- ---------------------------------------------------------------------------
+local function row_pull_cursor(parsed)
+    if parsed.synced_at then return parsed.synced_at end
+    local ts = 0
+    if parsed.updated_at and parsed.updated_at > ts then ts = parsed.updated_at end
+    if parsed.deleted_at and parsed.deleted_at > ts then ts = parsed.deleted_at end
+    return ts
+end
+M._row_pull_cursor = row_pull_cursor  -- exported for tests
+
 -- ---------------------------------------------------------------------------
 -- pushBook(book_row, opts, cb) — POST a single book row to /sync. Used
 -- after touchBook bumps updated_at on reader close, mirroring what
@@ -234,7 +253,7 @@ function M.pushChangedBooks(opts, cb)
         return
     end
 
-    local since = store:getLastPulledAt() or 0
+    local since = store:getLastPushedAt() or 0
     local changed = store:getChangedBooks(since)
     if #changed == 0 then
         logger.info("ReadestLibrary pushChangedBooks: nothing to push (since=" .. since .. ")")
@@ -272,7 +291,7 @@ function M.pushChangedBooks(opts, cb)
                     if cb then cb(false, body and body.error or "push failed", status) end
                     return
                 end
-                store:setLastPulledAt(max_ts)
+                store:setLastPushedAt(max_ts)
                 if cb then cb(true, #books_wire) end
             end)
     end)
@@ -381,7 +400,14 @@ function M.pullBooks(opts, cb)
             end
 
             local rows = body and body.books or {}
-            local max_ts = 0
+            -- Two separate watermarks, seeded from their stored values so a
+            -- partial/empty page never regresses either cursor (issue #4934):
+            --   pull_ts  — server synced_at; the `since` we send next pull.
+            --   push_ts  — local updated_at | deleted_at; getChangedBooks's
+            --              cursor, advanced here too so a just-pulled book
+            --              (already on the server) isn't re-pushed.
+            local pull_ts = opts.store:getLastPulledAt() or 0
+            local push_ts = opts.store:getLastPushedAt() or 0
             local upserted = 0
             for _, raw in ipairs(rows) do
                 local parsed = LibraryStore.parseSyncRow(raw)
@@ -389,19 +415,21 @@ function M.pullBooks(opts, cb)
                     parsed.user_id = opts.settings.user_id
                     opts.store:upsertBook(parsed)
                     upserted = upserted + 1
-                    -- Watermark = max of returned updated_at | deleted_at,
-                    -- not local now (codex round 1, finding 8).
-                    if parsed.updated_at and parsed.updated_at > max_ts then
-                        max_ts = parsed.updated_at
+                    local rc = row_pull_cursor(parsed)
+                    if rc > pull_ts then pull_ts = rc end
+                    if parsed.updated_at and parsed.updated_at > push_ts then
+                        push_ts = parsed.updated_at
                     end
-                    if parsed.deleted_at and parsed.deleted_at > max_ts then
-                        max_ts = parsed.deleted_at
+                    if parsed.deleted_at and parsed.deleted_at > push_ts then
+                        push_ts = parsed.deleted_at
                     end
                 end
             end
-            if max_ts > 0 then opts.store:setLastPulledAt(max_ts) end
+            opts.store:setLastPulledAt(pull_ts)
+            opts.store:setLastPushedAt(push_ts)
             logger.info("ReadestLibrary pullBooks complete: rows=" .. #rows
-                .. " upserted=" .. upserted .. " new_watermark=" .. max_ts)
+                .. " upserted=" .. upserted .. " new_watermark=" .. pull_ts
+                .. " push_watermark=" .. push_ts)
             if cb then cb(true, upserted) end
         end)
     end)

apps/readest.koplugin/spec/library/librarystore_spec.lua

@@ -89,9 +89,9 @@ describe("LibraryStore", function()
             s3:close()
         end)
 
-        it("sets PRAGMA user_version = 2", function()
+        it("sets PRAGMA user_version = 3", function()
             local store = LibraryStore.new({ user_id = "alice" })
-            assert.are.equal(2, store:getUserVersion())
+            assert.are.equal(3, store:getUserVersion())
             store:close()
         end)
 
@@ -109,7 +109,7 @@ describe("LibraryStore", function()
                 end
             end)
 
-            it("migrates an existing v1 DB: column added, row preserved, user_version=2", function()
+            it("migrates an existing v1 DB: column added, row preserved, user_version=current", function()
                 -- Build a temp file path; os.tmpname() creates the file, remove it
                 -- so sqlite.open() starts fresh instead of opening an existing file.
                 db_path = os.tmpname()
@@ -162,8 +162,9 @@ describe("LibraryStore", function()
                 -- Open the v1 file via LibraryStore; migration should run silently.
                 local store = LibraryStore.new({ user_id = "alice", db_path = db_path })
 
-                -- 1. user_version must now be 2
-                assert.are.equal(2, store:getUserVersion())
+                -- 1. Opening lands at the current schema (migrations are
+                -- cumulative: v1 -> v2 adds the column, v2 -> v3 splits cursors).
+                assert.are.equal(3, store:getUserVersion())
 
                 -- 2. The pre-existing row is still readable (no error, title intact)
                 local row = store:_getRowRaw("h-v1")
@@ -183,6 +184,75 @@ describe("LibraryStore", function()
                 store:close()
             end)
         end)
+
+        -- -----------------------------------------------------------------
+        -- v2 -> v3 migration: split the shared books cursor into a pull
+        -- cursor (server synced_at) and a push watermark (local updated_at).
+        -- The old shared `last_books_pulled_at` was advanced from client
+        -- updated_at, which a fast device clock could push into the future,
+        -- starving every later pull (issue #4934). The migration seeds the
+        -- new push watermark from the old value (so push dedup is unbroken)
+        -- and resets the pull cursor to 0, forcing one full re-pull that
+        -- re-establishes it on the server's synced_at clock.
+        -- -----------------------------------------------------------------
+        describe("v2 -> v3 migration", function()
+            local db_path
+            local SQ3 = require("lua-ljsqlite3/init")
+
+            after_each(function()
+                if db_path then
+                    os.remove(db_path)
+                    db_path = nil
+                end
+            end)
+
+            it("resets the pull cursor to 0 and seeds the push watermark from it", function()
+                db_path = os.tmpname()
+                os.remove(db_path)
+
+                -- Build a DB carrying a poisoned (future) shared cursor, then
+                -- force user_version back to 2 so reopening runs the migration.
+                local seed = LibraryStore.new({ user_id = "alice", db_path = db_path })
+                seed:setLastPulledAt(4102444800000)  -- ~2100-01-01, a future value
+                seed:close()
+                local raw = SQ3.open(db_path)
+                raw:exec("PRAGMA user_version = 2;")
+                raw:close()
+
+                local store = LibraryStore.new({ user_id = "alice", db_path = db_path })
+
+                assert.are.equal(3, store:getUserVersion())
+                -- Pull cursor reset → next sync does a full re-pull.
+                assert.are.equal(0, store:getLastPulledAt())
+                -- Push watermark seeded from the old shared cursor → no re-push storm.
+                assert.are.equal(4102444800000, store:getLastPushedAt())
+                store:close()
+            end)
+
+            it("is per-user (only migrates the rows it finds)", function()
+                db_path = os.tmpname()
+                os.remove(db_path)
+
+                local a = LibraryStore.new({ user_id = "alice", db_path = db_path })
+                a:setLastPulledAt(111)
+                a:close()
+                local b = LibraryStore.new({ user_id = "bob", db_path = db_path })
+                b:setLastPulledAt(222)
+                b:close()
+                local raw = SQ3.open(db_path)
+                raw:exec("PRAGMA user_version = 2;")
+                raw:close()
+
+                local alice = LibraryStore.new({ user_id = "alice", db_path = db_path })
+                assert.are.equal(0, alice:getLastPulledAt())
+                assert.are.equal(111, alice:getLastPushedAt())
+                alice:close()
+                local bob = LibraryStore.new({ user_id = "bob", db_path = db_path })
+                assert.are.equal(0, bob:getLastPulledAt())
+                assert.are.equal(222, bob:getLastPushedAt())
+                bob:close()
+            end)
+        end)
     end)
 
     -- =====================================================================
@@ -208,6 +278,18 @@ describe("LibraryStore", function()
             assert.are.equal(1781740800000, parsed.reading_status_updated_at)
             assert.are.equal("abandoned", parsed.reading_status)
         end)
+
+        it("parseSyncRow reads synced_at from the server ISO field (issue #4934)", function()
+            -- synced_at is the server-authoritative pull cursor. Even when a
+            -- book carries a future updated_at (a device with a fast clock),
+            -- synced_at reflects real server time.
+            local parsed = LibraryStore.parseSyncRow({
+                book_hash = "h3", title = "T",
+                updated_at = "2099-01-01T00:00:00+00:00",
+                synced_at  = "2026-06-18T00:00:00+00:00",
+            })
+            assert.are.equal(1781740800000, parsed.synced_at)
+        end)
     end)
 
     -- =====================================================================
@@ -950,6 +1032,17 @@ describe("LibraryStore", function()
             store:setLastPulledAt(T_RECENT)
             assert.are.equal(T_RECENT, store:getLastPulledAt())
         end)
+
+        it("getLastPushedAt returns nil before any setLastPushedAt", function()
+            assert.is_nil(store:getLastPushedAt())
+        end)
+
+        it("setLastPushedAt round-trips independently of the pull cursor", function()
+            store:setLastPulledAt(T_OLD)
+            store:setLastPushedAt(T_RECENT)
+            assert.are.equal(T_OLD, store:getLastPulledAt())
+            assert.are.equal(T_RECENT, store:getLastPushedAt())
+        end)
     end)
 
     -- =====================================================================

apps/readest.koplugin/spec/library/syncbooks_spec.lua

@@ -417,4 +417,89 @@ describe("library.syncbooks", function()
             end)
         end)
     end)
+
+    -- =====================================================================
+    -- _row_pull_cursor — per-row incremental-pull watermark contribution.
+    -- Mirrors computeMaxTimestamp at apps/readest-app/src/hooks/useSync.ts:29.
+    -- The server keys the books pull on the server-stamped `synced_at`
+    -- (issue #4678), so the cursor must track synced_at — NOT updated_at,
+    -- which is client-supplied and can run ahead of the server on a device
+    -- with a fast clock, starving every later pull (issue #4934).
+    -- =====================================================================
+    describe("_row_pull_cursor", function()
+        it("returns synced_at when present, ignoring a larger (future) updated_at", function()
+            assert.are.equal(100, syncbooks._row_pull_cursor({
+                synced_at = 100, updated_at = 999999999, deleted_at = nil,
+            }))
+        end)
+
+        it("falls back to max(updated_at, deleted_at) when synced_at is absent", function()
+            assert.are.equal(70, syncbooks._row_pull_cursor({ updated_at = 50, deleted_at = 70 }))
+            assert.are.equal(80, syncbooks._row_pull_cursor({ updated_at = 80 }))
+            assert.are.equal(90, syncbooks._row_pull_cursor({ deleted_at = 90 }))
+        end)
+
+        it("returns 0 for a row with no timestamps", function()
+            assert.are.equal(0, syncbooks._row_pull_cursor({}))
+        end)
+    end)
+
+    -- =====================================================================
+    -- pullBooks — end-to-end cursor advancement with an injected sync_auth +
+    -- client and a real in-memory LibraryStore. Regression guard for #4934:
+    -- the pull cursor must advance from synced_at, and the push watermark
+    -- (used for local-change detection) must advance from updated_at, so the
+    -- two never contaminate each other.
+    -- =====================================================================
+    describe("pullBooks cursor (issue #4934)", function()
+        local LibraryStore = require("library.librarystore")
+
+        local function fake_sync_auth(rows)
+            return {
+                withFreshToken = function(_self, _settings, _path, cb) cb(true) end,
+                getReadestSyncClient = function()
+                    return {
+                        pullBooks = function(_self2, _params, cb)
+                            cb(true, { books = rows }, 200)
+                        end,
+                    }
+                end,
+            }
+        end
+
+        local function pull(store, rows)
+            syncbooks.pullBooks({
+                sync_auth = fake_sync_auth(rows),
+                sync_path = "/tmp",
+                settings  = { user_id = "u1" },
+                store     = store,
+            }, function() end)
+        end
+
+        it("advances the pull cursor from synced_at, not a future updated_at", function()
+            local store = LibraryStore.new({ user_id = "u1" })
+            pull(store, {
+                { book_hash = "h1", title = "B",
+                  updated_at = "2099-01-01T00:00:00+00:00",   -- fast-clock device
+                  synced_at  = "2026-06-18T00:00:00+00:00" }, -- real server time
+            })
+            -- Pre-fix the cursor jumped to 2099 and the library never updated.
+            assert.are.equal(1781740800000, store:getLastPulledAt())  -- 2026-06-18Z
+            store:close()
+        end)
+
+        it("advances pull cursor (synced_at) and push watermark (updated_at) distinctly", function()
+            local store = LibraryStore.new({ user_id = "u1" })
+            pull(store, {
+                { book_hash = "h1", title = "B",
+                  updated_at = "2026-06-18T00:00:00+00:00",
+                  synced_at  = "2026-06-20T00:00:00+00:00" },
+            })
+            assert.are.equal(1781913600000, store:getLastPulledAt())  -- synced_at 2026-06-20Z
+            -- Push watermark tracks local updated_at so a just-pulled book is
+            -- not re-pushed on the next sync (dedup parity with the old cursor).
+            assert.are.equal(1781740800000, store:getLastPushedAt())  -- updated_at 2026-06-18Z
+            store:close()
+        end)
+    end)
 end)