diff --git a/README.md b/README.md index cbd1cda..e5354b8 100644 --- a/README.md +++ b/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 diff --git a/include/convex/paginated.h b/include/convex/paginated.h index e2885c7..3ffbf2f 100644 --- a/include/convex/paginated.h +++ b/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; diff --git a/integration/README.md b/integration/README.md index 24f19ba..fee53c5 100644 --- a/integration/README.md +++ b/integration/README.md @@ -143,6 +143,9 @@ curl -X POST https://.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`. diff --git a/integration/convex-test-project/convex/messages.ts b/integration/convex-test-project/convex/messages.ts index 32cf619..352c14d 100644 --- a/integration/convex-test-project/convex/messages.ts +++ b/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: { diff --git a/integration/tests/integration_test.cpp b/integration/tests/integration_test.cpp index 78c2e3b..11e65a7 100644 --- a/integration/tests/integration_test.cpp +++ b/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 #include #include #include @@ -269,6 +270,79 @@ TEST(Live, PaginatedQueryChainsPagesAndSeesLiveInserts) { (std::vector{"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 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 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{"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{"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 worst; + { + std::lock_guard lk(mu); + for (const paginated_snapshot& s : snaps) { + std::vector 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", {}); diff --git a/src/paginated.cpp b/src/paginated.cpp index ad30614..661a528 100644 --- a/src/paginated.cpp +++ b/src/paginated.cpp @@ -2,6 +2,7 @@ #include +#include #include #include #include @@ -26,7 +27,11 @@ struct page_fields { const value_array* items = nullptr; bool is_done = false; std::string continue_cursor; + /// Where the server suggests this page be cut in two. Absent when the + /// page is too small to have a split point. + std::optional split_cursor; bool split_required = false; + bool split_recommended = false; }; std::optional parse_page(const value& v) { @@ -42,9 +47,15 @@ std::optional parse_page(const value& v) { f.items = &page_it->second.as_array(); f.is_done = done_it->second.as_boolean(); f.continue_cursor = cursor_it->second.as_string(); - if (const auto status_it = o.find("pageStatus"); status_it != o.end()) { - f.split_required = - status_it->second.is_string() && status_it->second.as_string() == "SplitRequired"; + if (const auto split_it = o.find("splitCursor"); + split_it != o.end() && split_it->second.is_string()) { + f.split_cursor = split_it->second.as_string(); + } + if (const auto status_it = o.find("pageStatus"); + status_it != o.end() && status_it->second.is_string()) { + const std::string& status = status_it->second.as_string(); + f.split_required = status == "SplitRequired"; + f.split_recommended = status == "SplitRecommended"; } return f; } @@ -85,6 +96,10 @@ std::string_view pagination_status_name(pagination_status status) { } struct paginated_impl : std::enable_shared_from_this { + /// Identifies a page across reorderings. Page positions shift when a + /// split swaps one page for two, so callbacks address pages by key. + using page_key = std::uint64_t; + // Immutable after construction. client* owner; std::string udf_path; @@ -103,10 +118,31 @@ struct paginated_impl : std::enable_shared_from_this { bool stopped = false; struct page { + page_key key = 0; client::subscription sub; std::optional result; + /// Start of the page; absent means the beginning of the query. + std::optional cursor; + /// End of the page. Only set on the halves of a split, which pin + /// both ends of their range so the two of them cover exactly what + /// the page they replace covered. + std::optional end_cursor; }; + + /// A split in flight: two half-pages loading in parallel while the page + /// they will replace stays live and visible. Swapping them in only once + /// both have results keeps the combined list gapless at every moment. + struct pending_split { + page_key original = 0; + page first; + page second; + }; + + /// Active pages, in result order. std::vector pages; + /// Splits in flight. Their halves are not in `pages` yet. + std::vector splits; + page_key next_key = 0; paginated_impl(client& c, paginated_query::options&& o, paginated_query::snapshot_callback&& cb) @@ -117,46 +153,170 @@ struct paginated_impl : std::enable_shared_from_this { on_update(std::move(cb)), pagination_id(next_pagination_id()) {} - // Requires `m`. Appends and subscribes one page. - void add_page(std::size_t num_items, std::optional cursor) { + // Requires `m`. Looks up a page by key, in `pages` or in a split's halves. + page* find_page(page_key key) { + for (page& p : pages) { + if (p.key == key) return &p; + } + for (pending_split& s : splits) { + if (s.first.key == key) return &s.first; + if (s.second.key == key) return &s.second; + } + return nullptr; + } + + // Requires `m`. Index into `splits` of the split `key` is a half of. + std::optional split_of_half(page_key key) const { + for (std::size_t i = 0; i < splits.size(); ++i) { + if (splits[i].first.key == key || splits[i].second.key == key) return i; + } + return std::nullopt; + } + + // Requires `m`. True while `key` is being replaced by two halves. + bool is_splitting(page_key key) const { + return std::any_of(splits.begin(), splits.end(), + [key](const pending_split& s) { return s.original == key; }); + } + + // Requires `m`. Subscribes an already-stored page to its page query. + void subscribe_page(page& p, std::size_t num_items) { const std::uint64_t gen = generation; - const std::size_t index = pages.size(); - pages.emplace_back(); + const page_key key = p.key; value_object opts; // numItems and id are float64 on the wire: paginationOptsValidator // uses v.number(), which rejects Convex int64 ($integer). opts.emplace("numItems", value(static_cast(num_items))); - opts.emplace("cursor", cursor ? value(std::move(*cursor)) : value(nullptr)); + opts.emplace("cursor", p.cursor ? value(*p.cursor) : value(nullptr)); + if (p.end_cursor) opts.emplace("endCursor", value(*p.end_cursor)); opts.emplace("id", value(static_cast(pagination_id))); value_object args = user_args; args.insert_or_assign("paginationOpts", value(std::move(opts))); - pages[index].sub = owner->subscribe( + // subscribe() delivers an already-known result synchronously, so the + // page callback can run — and reset the session, destroying `p` — + // before this returns. Everything after it goes through find_page; + // if the page is gone, the local subscription drops on scope exit. + client::subscription sub = owner->subscribe( udf_path, std::move(args), - [self = shared_from_this(), gen, index](const function_result& r) { - self->on_page_result(gen, index, r); + [self = shared_from_this(), gen, key](const function_result& r) { + self->on_page_result(gen, key, r); }); + if (gen != generation) return; + if (page* current = find_page(key)) current->sub = std::move(sub); + } + + // Requires `m`. Appends and subscribes one page at the end of the list. + void add_page(std::size_t num_items, std::optional cursor) { + pages.emplace_back(); + page& p = pages.back(); + p.key = ++next_key; + p.cursor = std::move(cursor); + subscribe_page(p, num_items); } // Requires `m`. Drops every page and starts a fresh session. void do_reset() { ++generation; pagination_id = next_pagination_id(); + splits.clear(); pages.clear(); // unsubscribes add_page(initial_num_items, std::nullopt); } - void on_page_result(std::uint64_t gen, std::size_t index, const function_result& r) { + // Requires `m`. Starts replacing `original` with two halves covering the + // same range: (cursor, split_cursor] and (split_cursor, continue_cursor]. + void start_split(page_key original, const std::string& split_cursor, + const std::string& continue_cursor) { + const page* orig = find_page(original); + if (orig == nullptr) return; + pending_split s; + s.original = original; + s.first.key = ++next_key; + // The first half starts where the original page started — NOT at the + // beginning of the query. convex-js sent a null cursor here, which + // duplicated every item before the page (get-convex/convex-js#54981). + s.first.cursor = orig->cursor; + s.first.end_cursor = split_cursor; + s.second.key = ++next_key; + s.second.cursor = split_cursor; + s.second.end_cursor = continue_cursor; + const page_key first_key = s.first.key; + const page_key second_key = s.second.key; + splits.push_back(std::move(s)); + // Re-find between the two: subscribing the first half can re-enter + // and move the splits vector out from under us. + if (page* p = find_page(first_key)) subscribe_page(*p, initial_num_items); + if (page* p = find_page(second_key)) subscribe_page(*p, initial_num_items); + } + + // Requires `m`. Called when a half of split `index` gets a result. + void advance_split(std::size_t index) { + { + pending_split& s = splits[index]; + // Look for a failure before waiting on the peer. A split with a + // failed half can never complete, and while it sits here + // is_splitting blocks every later attempt to repair the page it + // was splitting — including an update that would have worked. + if ((s.first.result && !s.first.result->ok()) || + (s.second.result && !s.second.result->ok())) { + // An ordinary error; InvalidCursor resets before we get here. + // Drop the split and leave the page it was repairing in place. + // The next update of that page tries again. Retrying from here + // instead would re-subscribe the same failing query, whose + // cached error is delivered synchronously, and recurse. + splits.erase(splits.begin() + static_cast(index)); + return; + } + if (!s.first.result || !s.second.result) return; + } + pending_split done = std::move(splits[index]); + splits.erase(splits.begin() + static_cast(index)); + const auto it = std::find_if(pages.begin(), pages.end(), [&](const page& p) { + return p.key == done.original; + }); + if (it == pages.end()) return; + const auto at = it - pages.begin(); + const page_key first_key = done.first.key; + const page_key second_key = done.second.key; + *it = std::move(done.first); // unsubscribes the page being replaced + pages.insert(pages.begin() + at + 1, std::move(done.second)); + // A half can be oversized in its own right (the range it inherited + // may still be too big), so keep splitting until it isn't. + consider_split(first_key); + consider_split(second_key); + } + + // Requires `m`. Splits `key` if the server asked for it, or if the page + // has outgrown twice the requested size. Mirrors convex-js's + // processPaginatedQuerySplits. + void consider_split(page_key key) { + page* p = find_page(key); + if (p == nullptr || !p->result || !p->result->ok()) return; + if (is_splitting(key)) return; + const auto f = parse_page(p->result->get_value()); + if (!f) return; + // No split point means nothing to do: an incomplete page stays out of + // the snapshot, and its subscription is live, so a later result that + // fits the limits repairs it on its own. + if (!f->split_cursor) return; + if (f->split_required || f->split_recommended || + f->items->size() > initial_num_items * 2) { + start_split(key, *f->split_cursor, f->continue_cursor); + } + } + + void on_page_result(std::uint64_t gen, page_key key, const function_result& r) { std::lock_guard lk(m); if (stopped || gen != generation) return; - pages[index].result = r; + page* p = find_page(key); + if (p == nullptr) return; + p->result = r; if (is_invalid_cursor(r)) { do_reset(); - } else if (r.ok()) { - if (const auto f = parse_page(r.get_value()); f && f->split_required) { - // The page outgrew the server's read limits and may be - // incomplete. convex-js splits it; we reset (see header). - do_reset(); - } + } else if (const auto index = split_of_half(key); index) { + advance_split(*index); + } else { + consider_split(key); } notify(); } @@ -184,12 +344,21 @@ struct paginated_impl : std::enable_shared_from_this { "\" returned a value that is not a PaginationResult"); return snap; } + if (f->split_required) { + // The server could not read this page in full, so its items + // may be missing part of the range. Stop before it rather + // than publish a list with a hole in it — the split (or the + // reset) that repairs it is already under way. This is what + // convex-js's usePaginatedQuery does. + all_loaded = false; + break; + } snap.results.insert(snap.results.end(), f->items->begin(), f->items->end()); last_is_done = f->is_done; } if (!all_loaded) { - snap.status = (pages.size() == 1) ? pagination_status::loading_first_page - : pagination_status::loading_more; + snap.status = snap.results.empty() ? pagination_status::loading_first_page + : pagination_status::loading_more; } else { snap.status = last_is_done ? pagination_status::exhausted : pagination_status::can_load_more; @@ -209,6 +378,7 @@ struct paginated_impl : std::enable_shared_from_this { std::lock_guard lk(m); stopped = true; on_update = nullptr; + splits.clear(); pages.clear(); } }; diff --git a/tests/paginated_test.cpp b/tests/paginated_test.cpp index e157173..d1a751d 100644 --- a/tests/paginated_test.cpp +++ b/tests/paginated_test.cpp @@ -1,6 +1,6 @@ // paginated_query tests against the mock transport: paginationOpts wire // shape, cursor chaining, combined results across live page subscriptions, -// status transitions, args-change / InvalidCursor / SplitRequired resets, +// status transitions, args-change / InvalidCursor resets, page splitting, // and journal resend on reconnect (the seam-stability guarantee). #include @@ -50,9 +50,10 @@ constexpr const char* TS4 = "BAAAAAAAAAA="; // A PaginationResult value: string items for easy comparison. json page_value(const std::vector& items, bool is_done, const std::string& cursor, - const char* page_status = nullptr) { + const char* page_status = nullptr, const char* split_cursor = nullptr) { json v{{"page", items}, {"isDone", is_done}, {"continueCursor", cursor}}; if (page_status) v["pageStatus"] = page_status; + if (split_cursor) v["splitCursor"] = split_cursor; return v; } @@ -115,9 +116,41 @@ struct harness { json sent_json(std::size_t attempt, std::size_t index) { return json::parse(transport->sent(attempt).at(index)); } + // The Add modification for `query_id`, from anywhere in the sent stream + // (a split adds two pages, so positions are no longer predictable). + json add_for(int query_id, std::size_t attempt = 0) { + for (const std::string& raw : transport->sent(attempt)) { + const json j = json::parse(raw); + if (j.value("type", "") != "ModifyQuerySet") continue; + for (const json& mod : j["modifications"]) { + if (mod["type"] == "Add" && mod["queryId"] == query_id) return mod; + } + } + return json::object(); + } + bool sent_remove(int query_id, std::size_t attempt = 0) { + for (const std::string& raw : transport->sent(attempt)) { + const json j = json::parse(raw); + if (j.value("type", "") != "ModifyQuerySet") continue; + for (const json& mod : j["modifications"]) { + if (mod["type"] == "Remove" && mod["queryId"] == query_id) return true; + } + } + return false; + } + // A copy of the paginationOpts a page was subscribed with. Returned by + // value: a reference into add_for's temporary would dangle. + json opts_for(int query_id, std::size_t attempt = 0) { + const json add = add_for(query_id, attempt); + return add.contains("args") ? add["args"][0]["paginationOpts"] : json::object(); + } + double pagination_id(int query_id) { return opts_for(query_id)["id"].get(); } bool wait_status(pagination_status s) { return pump_until(c, [&] { return pq.snapshot().status == s; }); } + // Pump with nothing to wait for, so anything pending lands before an + // assertion that something did *not* happen. + void settle() { pump_until(c, [] { return false; }, 50ms); } }; } // namespace @@ -360,36 +393,288 @@ TEST(Paginated, OtherErrorsSurfaceWithoutReset) { EXPECT_EQ(h.transport->sent(0).size(), 3u) << "no reset traffic for ordinary errors"; } -TEST(Paginated, SplitRequiredResetsPagination) { +TEST(Paginated, SplitRequiredSplitsPageInTwo) { harness h; + // A page that outgrew the read limits: incomplete, with a split point. h.transport->server_send( 0, transition_page({0, TS0}, {1, TS1}, 0, - page_value({"a", "b", "c", "d"}, false, "c1", "SplitRequired"), "j0")); - // v1 has no page splitting: an incomplete (SplitRequired) page triggers - // a reset instead, which re-fetches right-sized pages. The reset runs in - // the pumped page callback, so pump while waiting for + Remove(q0), Add(q1). + page_value({"a", "b", "c", "d"}, false, "c1", "SplitRequired", "s1"), + "j0")); + // Two halves are subscribed; the split runs in the pumped page callback. ASSERT_TRUE(pump_until(h.c, [&] { return h.transport->sent(0).size() >= 4; })); + + const json first_opts = h.opts_for(1); + const json second_opts = h.opts_for(2); + ASSERT_FALSE(first_opts.empty()); + ASSERT_FALSE(second_opts.empty()); + // (beginning, s1] and (s1, c1]: together exactly the original range. + EXPECT_TRUE(first_opts["cursor"].is_null()); + EXPECT_EQ(first_opts["endCursor"], "s1"); + EXPECT_EQ(second_opts["cursor"], "s1"); + EXPECT_EQ(second_opts["endCursor"], "c1"); + EXPECT_EQ(first_opts["numItems"].get(), 2.0); + EXPECT_EQ(first_opts["id"].get(), h.pagination_id(0)) + << "a split stays inside the same pagination session"; + + // The page is incomplete, so it is not published while the split runs: + // a short list beats a list with a hole in it. + EXPECT_TRUE(h.pq.snapshot().results.empty()); EXPECT_EQ(h.pq.snapshot().status, pagination_status::loading_first_page); - const json add_mqs = h.sent_json(0, 3); - const json& add = add_mqs["modifications"][0]; - EXPECT_EQ(add["type"], "Add"); - EXPECT_EQ(add["queryId"], 1); - EXPECT_TRUE(add["args"][0]["paginationOpts"]["cursor"].is_null()); + EXPECT_FALSE(h.sent_remove(0)) << "the original page stays subscribed"; h.transport->server_send( - 0, transition_page({1, TS1}, {3, TS2}, 1, page_value({"a", "b"}, false, "e1"), "j1")); + 0, transition_page({1, TS1}, {2, TS2}, 1, page_value({"a", "b"}, false, "s1"), "j1")); + h.settle(); + EXPECT_TRUE(h.pq.snapshot().results.empty()) + << "one half is not a page: no partial swap"; + EXPECT_FALSE(h.sent_remove(0)); + + h.transport->server_send( + 0, transition_page({2, TS2}, {3, TS3}, 2, page_value({"c", "d"}, false, "c1"), "j2")); + ASSERT_TRUE(pump_until(h.c, [&] { return h.sent_remove(0); })); + const auto snap = h.pq.snapshot(); + EXPECT_EQ(snap.results, items({"a", "b", "c", "d"})) << "same items, now over two pages"; + EXPECT_EQ(snap.status, pagination_status::can_load_more); + + // The list keeps growing from the second half's cursor. + ASSERT_TRUE(h.pq.load_more(2)); + EXPECT_EQ(h.opts_for(3)["cursor"], "c1"); +} + +TEST(Paginated, IncompletePageHidesItselfButNotThePagesBeforeIt) { + // Everything up to the incomplete page keeps showing; the truncated page + // and anything after it wait for the split. Mirrors convex-js's + // usePaginatedQuery, which stops results before a SplitRequired page. + harness h; + h.transport->server_send(0, transition_page({0, TS0}, {1, TS1}, 0, + page_value({"a", "b"}, false, "c1"), "j0")); ASSERT_TRUE(h.wait_status(pagination_status::can_load_more)); - EXPECT_EQ(h.pq.snapshot().results, items({"a", "b"})); + ASSERT_TRUE(h.pq.load_more(2)); + + // Page two comes back truncated: the server read past its limit, so "c" + // may not be all of (c1, c2]. + h.transport->server_send( + 0, transition_page({1, TS1}, {2, TS2}, 1, + page_value({"c"}, false, "c2", "SplitRequired", "s2"), "j1")); + ASSERT_TRUE(pump_until(h.c, [&] { return !h.add_for(3).empty(); })); + const auto during = h.pq.snapshot(); + EXPECT_EQ(during.results, items({"a", "b"})) << "page one is complete and stays"; + EXPECT_EQ(during.status, pagination_status::loading_more); + EXPECT_TRUE(during.is_loading()); + + // Both halves land: the full range appears at once. + h.transport->server_send( + 0, transition_page({2, TS2}, {3, TS3}, 2, page_value({"c"}, false, "s2"), "j2")); + h.transport->server_send( + 0, transition_page({3, TS3}, {4, TS4}, 3, page_value({"d"}, true, "c2"), "j3")); + ASSERT_TRUE(h.wait_status(pagination_status::exhausted)); + EXPECT_EQ(h.pq.snapshot().results, items({"a", "b", "c", "d"})); } -TEST(Paginated, SplitRecommendedIsIgnored) { +TEST(Paginated, SplitFirstHalfStartsAtTheOriginalPagesCursor) { + // Regression guard for get-convex/convex-js#54981: convex-js sent a null + // cursor for the first half, so splitting any page but the first re-read + // the list from the beginning and duplicated everything before it. + harness h; + h.transport->server_send(0, transition_page({0, TS0}, {1, TS1}, 0, + page_value({"a", "b"}, false, "c1"), "j0")); + ASSERT_TRUE(h.wait_status(pagination_status::can_load_more)); + ASSERT_TRUE(h.pq.load_more(2)); + ASSERT_TRUE(h.transport->wait_for_sent(0, 3)); + + // The second page (which starts at c1) grew and needs splitting. + h.transport->server_send( + 0, transition_page({1, TS1}, {2, TS2}, 1, + page_value({"c", "d", "e"}, false, "c2", "SplitRecommended", "s2"), + "j1")); + ASSERT_TRUE(pump_until(h.c, [&] { return h.transport->sent(0).size() >= 5; })); + + const json first_opts = h.opts_for(2); + EXPECT_FALSE(first_opts["cursor"].is_null()) << "must not restart from the beginning"; + EXPECT_EQ(first_opts["cursor"], "c1"); + EXPECT_EQ(first_opts["endCursor"], "s2"); + EXPECT_EQ(h.opts_for(3)["cursor"], "s2"); + + h.transport->server_send( + 0, transition_page({2, TS2}, {3, TS3}, 2, page_value({"c", "d"}, false, "s2"), "j2")); + h.transport->server_send( + 0, transition_page({3, TS3}, {4, TS4}, 3, page_value({"e"}, true, "c2"), "j3")); + ASSERT_TRUE(h.wait_status(pagination_status::exhausted)); + EXPECT_EQ(h.pq.snapshot().results, items({"a", "b", "c", "d", "e"})); +} + +TEST(Paginated, OversizedPageSplitsWithoutServerAsking) { + // convex-js splits any page that grew past twice the requested size, + // whatever pageStatus says. + harness h; // initial_num_items = 2 + h.transport->server_send( + 0, transition_page({0, TS0}, {1, TS1}, 0, + page_value({"a", "b", "c", "d", "e"}, false, "c1", nullptr, "s1"), + "j0")); + ASSERT_TRUE(pump_until(h.c, [&] { return h.transport->sent(0).size() >= 4; })); + EXPECT_EQ(h.opts_for(1)["endCursor"], "s1"); + EXPECT_EQ(h.opts_for(2)["cursor"], "s1"); +} + +TEST(Paginated, PageWithinSizeAndNoSplitCursorIsLeftAlone) { harness h; h.transport->server_send( 0, transition_page({0, TS0}, {1, TS1}, 0, page_value({"a", "b", "c"}, false, "c1", "SplitRecommended"), "j0")); ASSERT_TRUE(h.wait_status(pagination_status::can_load_more)); EXPECT_EQ(h.pq.snapshot().results, items({"a", "b", "c"})); - EXPECT_EQ(h.transport->sent(0).size(), 2u) << "no reset, no extra traffic"; + EXPECT_EQ(h.transport->sent(0).size(), 2u) << "nothing to split on: no extra traffic"; +} + +TEST(Paginated, IncompletePageWithNoSplitPointWaitsInsteadOfResetting) { + // The server says the page is incomplete but gives nothing to split on. + // There is no client-side repair for that, so the page stays out of the + // snapshot and its live subscription waits. Re-fetching the whole session + // to force the issue would be a loop with nothing to bound it. + harness h; + h.transport->server_send( + 0, transition_page({0, TS0}, {1, TS1}, 0, + page_value({"a"}, false, "c1", "SplitRequired"), "j0")); + h.settle(); + EXPECT_EQ(h.pq.snapshot().status, pagination_status::loading_first_page); + EXPECT_TRUE(h.pq.snapshot().results.empty()); + EXPECT_FALSE(h.pq.snapshot().error.has_value()) << "loading, not an error"; + EXPECT_EQ(h.transport->sent(0).size(), 2u) << "no reset traffic at all"; + EXPECT_FALSE(h.sent_remove(0)) << "the subscription stays, so it can recover"; + + // It recovers on its own when the server can read the page in full. + h.transport->server_send(0, transition_page({1, TS1}, {2, TS2}, 0, + page_value({"a", "b"}, true, "c1"), "j0b")); + ASSERT_TRUE(h.wait_status(pagination_status::exhausted)); + EXPECT_EQ(h.pq.snapshot().results, items({"a", "b"})); +} + +TEST(Paginated, SplitIsAbandonedAsSoonAsAHalfFails) { + // The peer may never arrive. Waiting for it before dropping a split that + // cannot complete leaves the page it was repairing blocked, because + // consider_split skips a page while a split of it is pending. + harness h; + h.transport->server_send( + 0, transition_page({0, TS0}, {1, TS1}, 0, + page_value({"a", "b", "c", "d"}, false, "c1", "SplitRequired", "s1"), + "j0")); + ASSERT_TRUE(pump_until(h.c, [&] { return !h.add_for(2).empty(); })); + EXPECT_TRUE(h.pq.snapshot().results.empty()) << "incomplete page stays hidden"; + + // One half fails; the other never returns. + h.transport->server_send( + 0, transition_failed({1, TS1}, {2, TS2}, 1, "Server Error: Uncaught Error: boom")); + ASSERT_TRUE(pump_until(h.c, [&] { return h.sent_remove(1) && h.sent_remove(2); })) + << "both halves drop without waiting for the peer"; + EXPECT_EQ(h.pq.snapshot().status, pagination_status::loading_first_page); + EXPECT_FALSE(h.sent_remove(0)) << "the page itself stays subscribed"; + + // The page is no longer blocked: its next update starts a fresh split. + h.transport->server_send( + 0, transition_page({2, TS2}, {3, TS3}, 0, + page_value({"a", "b", "c", "d"}, false, "c1", "SplitRequired", "s1"), + "j0b")); + ASSERT_TRUE(pump_until(h.c, [&] { return !h.add_for(4).empty(); })) + << "a dead split must not block the retry"; + EXPECT_EQ(h.opts_for(3)["endCursor"], "s1"); + EXPECT_EQ(h.opts_for(4)["cursor"], "s1"); + + // And that one completes. + h.transport->server_send( + 0, transition_page({3, TS3}, {4, TS4}, 3, page_value({"a", "b"}, false, "s1"), "k0")); + h.transport->server_send( + 0, transition_page({4, TS4}, {5, TS4}, 4, page_value({"c", "d"}, true, "c1"), "k1")); + ASSERT_TRUE(h.wait_status(pagination_status::exhausted)); + EXPECT_EQ(h.pq.snapshot().results, items({"a", "b", "c", "d"})); +} + +TEST(Paginated, FailedSplitLeavesThePageAndRetriesOnTheNextUpdate) { + // A half failing is not fatal: drop the split, keep the page, and try + // again when the server next sends that page. One attempt per transition, + // never a loop the client drives. + harness h; + h.transport->server_send( + 0, transition_page({0, TS0}, {1, TS1}, 0, + page_value({"a", "b", "c", "d", "e"}, false, "c1", nullptr, "s1"), + "j0")); + ASSERT_TRUE(pump_until(h.c, [&] { return !h.add_for(2).empty(); })); + h.transport->server_send( + 0, transition_page({1, TS1}, {2, TS2}, 1, page_value({"a", "b"}, false, "s1"), "j1")); + h.transport->server_send( + 0, transition_failed({2, TS2}, {3, TS3}, 2, "Server Error: Uncaught Error: boom")); + ASSERT_TRUE(pump_until(h.c, [&] { return h.sent_remove(1) && h.sent_remove(2); })); + + const auto snap = h.pq.snapshot(); + EXPECT_EQ(snap.status, pagination_status::can_load_more) << "a failed split is not an error"; + EXPECT_EQ(snap.results, items({"a", "b", "c", "d", "e"})); + EXPECT_FALSE(h.sent_remove(0)); + + // The next update of the same page starts a fresh split attempt. + h.transport->server_send( + 0, transition_page({3, TS3}, {4, TS4}, 0, + page_value({"a", "b", "c", "d", "e", "f"}, false, "c1", nullptr, "s1"), + "j0b")); + ASSERT_TRUE(pump_until(h.c, [&] { return !h.add_for(4).empty(); })) + << "the split is retried, not disabled forever"; + EXPECT_EQ(h.opts_for(3)["endCursor"], "s1"); + EXPECT_EQ(h.opts_for(4)["cursor"], "s1"); +} + +TEST(Paginated, InvalidCursorInASplitHalfResetsPagination) { + harness h; + h.transport->server_send( + 0, transition_page({0, TS0}, {1, TS1}, 0, + page_value({"a", "b", "c", "d"}, false, "c1", "SplitRequired", "s1"), + "j0")); + ASSERT_TRUE(pump_until(h.c, [&] { return !h.add_for(2).empty(); })); + + h.transport->server_send( + 0, transition_failed({1, TS1}, {2, TS2}, 1, + "InvalidCursor: Tried to run a query starting from a cursor " + "created by a different query")); + ASSERT_TRUE(h.wait_status(pagination_status::loading_first_page)); + EXPECT_TRUE(h.pq.snapshot().results.empty()); + ASSERT_TRUE(pump_until(h.c, [&] { return !h.add_for(3).empty(); })); + const json opts = h.opts_for(3); + EXPECT_TRUE(opts["cursor"].is_null()); + EXPECT_FALSE(opts.contains("endCursor")); + EXPECT_NE(opts["id"].get(), h.pagination_id(0)); +} + +TEST(Paginated, SplitPagesResubscribeWithTheirEndCursorsAfterReconnect) { + harness h; + h.transport->server_send( + 0, transition_page({0, TS0}, {1, TS1}, 0, + page_value({"a", "b", "c", "d"}, false, "c1", "SplitRequired", "s1"), + "j0")); + ASSERT_TRUE(pump_until(h.c, [&] { return !h.add_for(2).empty(); })); + h.transport->server_send( + 0, transition_page({1, TS1}, {2, TS2}, 1, page_value({"a", "b"}, false, "s1"), "j1")); + h.transport->server_send( + 0, transition_page({2, TS2}, {3, TS3}, 2, page_value({"c", "d"}, false, "c1"), "j2")); + ASSERT_TRUE(pump_until(h.c, [&] { return h.sent_remove(0); })); + + h.transport->server_close(0, "SimulatedNetworkBlip"); + ASSERT_TRUE(h.transport->wait_for_attempts(2)); + h.transport->open(1); + ASSERT_TRUE(h.transport->wait_for_sent(1, 2)); + + const json mqs = h.sent_json(1, 1); + ASSERT_EQ(mqs["modifications"].size(), 2u) << "only the two halves are active"; + for (const json& mod : mqs["modifications"]) { + const json& opts = mod["args"][0]["paginationOpts"]; + ASSERT_TRUE(opts.contains("endCursor")) << "split bounds survive a reconnect"; + if (mod["queryId"] == 1) { + EXPECT_TRUE(opts["cursor"].is_null()); + EXPECT_EQ(opts["endCursor"], "s1"); + EXPECT_EQ(mod["journal"], "j1"); + } else { + EXPECT_EQ(mod["queryId"], 2); + EXPECT_EQ(opts["cursor"], "s1"); + EXPECT_EQ(opts["endCursor"], "c1"); + EXPECT_EQ(mod["journal"], "j2"); + } + } } TEST(Paginated, MalformedPaginationResultIsAnError) {