Split oversized pagination pages instead of resetting

README.md

@@ -17,8 +17,10 @@ On top of it, `convex::paginated_query` implements the convex-js
 `usePaginatedQuery` state machine as a growing, live-updating list over a
 paginated query. It keeps one live subscription per page, exposes
 `load_more`, avoids seams at page boundaries via query journals, and resets
-on argument changes or stale cursors. For one-shot calls without a
-WebSocket there is `convex::http_client`, plus file storage
+on argument changes or stale cursors. A page that grows past twice the
+requested size is split in two, the same way convex-js does it, so a busy
+list does not drift into the server's read limits. For one-shot calls
+without a WebSocket there is `convex::http_client`, plus file storage
 upload/download helpers.
 
 The protocol core is sans-IO: `convex::base_client` does no networking at

include/convex/paginated.h

@@ -11,12 +11,29 @@
 // (paginationOptsValidator) and return a PaginationResult:
 //   { page: [...], isDone: bool, continueCursor: string }.
 //
-// Divergence from convex-js (v1): page splitting is NOT implemented. When
-// the server reports pageStatus == "SplitRequired" (a page grew past the
-// read limits and may be incomplete), convex-js splits the page in two via
-// splitCursor; this helper instead resets pagination, which re-fetches
-// correctly-sized pages. "SplitRecommended" is ignored. Revisit if reset
-// churn ever matters in practice.
+// Page splitting. A page's range is fixed by its journal, so inserts into
+// that range make the page grow; left alone it eventually outgrows the
+// server's read limits and comes back incomplete. Like convex-js, this
+// helper splits such a page in two at the server-supplied splitCursor —
+// subscribing both halves with explicit cursor/endCursor bounds and swapping
+// them in for the original only once both have loaded, so the combined list
+// is gapless and duplicate-free at every moment. A page is split when the
+// server reports pageStatus "SplitRequired" or "SplitRecommended", or when
+// it has grown past twice initial_num_items.
+//
+// A page the server reports as SplitRequired may be missing part of its
+// range, so the snapshot stops before it: the pages ahead of it stay visible
+// and the status goes back to loading until a split repairs it. A list with a
+// hole in it is worse than a shorter list. convex-js's usePaginatedQuery does
+// the same.
+//
+// Nothing here forces the issue beyond that. If the server reports a page
+// incomplete but gives no splitCursor to split it on, the page simply stays
+// out of the snapshot; its subscription is live, so a later result that fits
+// the read limits repairs it. A query heavy enough that no page ever fits
+// leaves the list loading indefinitely — again matching convex-js, which has
+// no recovery path for that case either. The alternative, re-fetching the
+// whole session to force the issue, is a loop with nothing to bound it.
 //
 // Threading. All callbacks (page updates) arrive through the owning client's
 // delivery mechanism — the process_events() pump by default. The on_update
@@ -87,8 +104,9 @@ struct paginated_impl;
 ///
 /// Pagination resets — dropping every page and re-subscribing a fresh first
 /// page with a new cache-buster id — happen when set_args() changes the
-/// arguments, when a page fails with InvalidCursor (cursors went stale), or
-/// when the server reports a page as SplitRequired (see file comment).
+/// arguments, when reset() is called, and when a page fails with InvalidCursor
+/// (cursors went stale). Splitting an oversized page is not a reset: loaded
+/// pages stay put.
 class paginated_query {
 public:
     using snapshot_callback = std::function<void(const paginated_snapshot&)>;

integration/README.md

@@ -143,6 +143,9 @@ curl -X POST https://<your-deployment>.convex.cloud/api/query \
 - `messages:list` (query): args `{channel}`, returns messages oldest-first.
 - `messages:listPaginated` (query): args `{channel, paginationOpts}`, returns
   a `PaginationResult` page of messages oldest-first (for `paginated_query`).
+- `messages:listPaginatedDesc` (query): the same, newest-first. Inserts land
+  inside the first page's range, so this is the one that grows a page past
+  its requested size and makes `paginated_query` split it.
 - `messages:send` (mutation): args `{channel, author, body}`, returns the new id.
 - `messages:clearAll` (mutation): no args, deletes all, returns count removed.
 - `counters:get` (query): args `{name}`, returns `number | null`.

integration/convex-test-project/convex/messages.ts

@@ -27,6 +27,20 @@ export const listPaginated = query({
   },
 });
 
+// Paginated list of messages, newest first. Inserts land inside the first
+// page's cursor range instead of after the last page, so this is the query
+// the paginated_query page-splitting test grows a page with.
+export const listPaginatedDesc = query({
+  args: { channel: v.string(), paginationOpts: paginationOptsValidator },
+  handler: async (ctx, { channel, paginationOpts }) => {
+    return await ctx.db
+      .query("messages")
+      .withIndex("by_channel", (q) => q.eq("channel", channel))
+      .order("desc")
+      .paginate(paginationOpts);
+  },
+});
+
 // Send (insert) a new message. Returns the new document id.
 export const send = mutation({
   args: {

integration/tests/integration_test.cpp

@@ -8,6 +8,7 @@
 //   CONVEX_CLOUD_URL       enables the TLS smoke test against a cloud dev
 //                          deployment when set (wss://)
 
+#include <algorithm>
 #include <atomic>
 #include <chrono>
 #include <cmath>
@@ -269,6 +270,79 @@ TEST(Live, PaginatedQueryChainsPagesAndSeesLiveInserts) {
               (std::vector<std::string>{"m1", "m2", "m3", "m4", "m5", "m6"}));
 }
 
+TEST(Live, PaginatedQuerySplitsAGrowingPage) {
+    // Newest-first, so every insert lands inside the first page's cursor
+    // range rather than after the last page. That is how a page outgrows the
+    // size it was requested at, and what page splitting exists to handle.
+    auto c = make_client(local_url());
+    const std::string channel = unique_name("split");
+
+    auto send = [&](const std::string& body) {
+        auto m = c.mutation("messages:send", {{"channel", value(channel)},
+                                              {"author", value("cpp")},
+                                              {"body", value(body)}});
+        ASSERT_TRUE(await(m).ok());
+    };
+    for (int i = 1; i <= 3; ++i) send("m" + std::to_string(i));
+
+    std::mutex mu;
+    std::condition_variable cv;
+    std::vector<paginated_snapshot> snaps;
+    paginated_query pq(
+        c, {"messages:listPaginatedDesc", {{"channel", value(channel)}}, /*initial_num_items=*/2},
+        [&](const paginated_snapshot& s) {
+            std::lock_guard lk(mu);
+            snaps.push_back(s);
+            cv.notify_all();
+        });
+    auto wait_snap = [&](auto pred) {
+        std::unique_lock lk(mu);
+        return cv.wait_for(lk, 15s, [&] { return !snaps.empty() && pred(snaps.back()); });
+    };
+    auto bodies = [](const paginated_snapshot& s) {
+        std::vector<std::string> out;
+        for (const value& doc : s.results) out.push_back(doc.as_object().at("body").as_string());
+        return out;
+    };
+
+    ASSERT_TRUE(wait_snap([](const paginated_snapshot& s) {
+        return s.status == pagination_status::can_load_more && s.results.size() == 2;
+    }));
+    ASSERT_TRUE(pq.load_more(2));
+    ASSERT_TRUE(wait_snap(
+        [](const paginated_snapshot& s) { return s.status == pagination_status::exhausted; }));
+    EXPECT_EQ(bodies(pq.snapshot()), (std::vector<std::string>{"m3", "m2", "m1"}));
+
+    // Three more inserts push the first page from 2 items to 5, past twice
+    // the size it was requested at. The helper splits it against the real
+    // server's splitCursor; the list must stay whole and in order through it.
+    for (int i = 4; i <= 6; ++i) send("m" + std::to_string(i));
+    ASSERT_TRUE(wait_snap([](const paginated_snapshot& s) { return s.results.size() == 6; }));
+    const auto after_split = pq.snapshot();
+    EXPECT_EQ(after_split.status, pagination_status::exhausted);
+    EXPECT_EQ(bodies(after_split),
+              (std::vector<std::string>{"m6", "m5", "m4", "m3", "m2", "m1"}));
+
+    // No snapshot along the way may repeat an item: the halves replace the
+    // page they split only once both have loaded.
+    std::vector<std::string> worst;
+    {
+        std::lock_guard lk(mu);
+        for (const paginated_snapshot& s : snaps) {
+            std::vector<std::string> seen = bodies(s);
+            std::sort(seen.begin(), seen.end());
+            const auto dup = std::adjacent_find(seen.begin(), seen.end());
+            if (dup != seen.end()) worst = bodies(s);
+        }
+    }
+    EXPECT_TRUE(worst.empty()) << "a snapshot showed an item twice: " << testing::PrintToString(worst);
+
+    // The list still grows from the last page after the split.
+    send("m7");
+    ASSERT_TRUE(wait_snap([](const paginated_snapshot& s) { return s.results.size() == 7; }));
+    EXPECT_EQ(bodies(pq.snapshot()).front(), "m7");
+}
+
 TEST(Live, ConvexErrorCarriesData) {
     auto c = make_client(local_url());
     auto future = c.query("errors:throwConvexError", {});