feat: limit/offset on cog_gov_search() and cog_balances() (#57)

verbs were left materializing everything and slicing in R -- the pattern behind
the 2026-08-06 production incident. cog_gov_search() had no LIMIT at all, so an
unfiltered call returns the entire 40,336-row crosswalk.

Extracted the #39 machinery into R/pagination.R first (.validate_pagination(),
.paginate_sql(), .take_pagination_total()) rather than growing a third inline
copy: three definitions of what total_rows means is three places for it to
drift. Conflict refusals stay at the call sites because each verb's conflict
set differs. .verb_spendrev() now uses the shared helpers and is unchanged in
behaviour.

The empty-page fallback query is now passed as a thunk, so the unpaginated SQL
is only BUILT when an offset actually lands past the end instead of on every
paged call.

Two things #57 did not anticipate:

- cog_gov_search()'s ORDER BY was not a total order. population_acs DESC NULLS
  LAST leaves ties -- and the whole NULL block -- in scan order, so two requests
  can order them differently and a paged sweep duplicates one row while dropping
  another. Added canonical_govid as tiebreaker. Unpaginated output changes only
  in the relative order of already-tied rows.

- Basket mode returns one resolved row per requested name plus a sidecar
  covering all of them, so a page of it is not a page of anything the caller
  asked for. Refused with uscogdata_basket_pagination_conflict rather than
  silently ignoring the arguments.

Both default to NULL, so cog-api adopts them behind its existing formals()
probe with no lockstep deploy.

Suite: 1067 passed, 0 failed, 0 warnings (2 pre-existing live-corpus skips).

R/balances.R

@@ -47,6 +47,12 @@
 #' @param recipe Optional harmonization recipe id (see [cog_recipes()]).
 #'   `"cash_securities_z77_wide"` and `"cash_securities_z78_wide"` bridge the
 #'   wide era to the modern one.
+#' @param limit Maximum number of result rows to return, pushed into the SQL
+#'   rather than applied after materializing every row. `NULL` (default)
+#'   returns everything. Cannot be combined with `recipe` -- see `offset` and
+#'   `total_rows`.
+#' @param offset Rows to skip before `limit` starts counting (0-based).
+#'   Ignored if `limit` is `NULL`; defaults to `0L` when `limit` is set.
 #'
 #' @return Tibble with columns `year`, `canonical_govid`, `gov_name`,
 #'   `balance_subtype`, `category`, `amt_nominal`, `codes_included`,
@@ -64,11 +70,16 @@
 #'   and `truncated` (the observed subtypes whose coverage falls short of the
 #'   requested years). `expenditure_concept`/`revenue_concept` are `NA` --
 #'   holdings are a stock, not a flow, so neither concept vocabulary applies.
+#'
+#'   When `limit` is set, also carries a `total_rows` attribute: the full
+#'   unpaginated row count, computed by the same query (`COUNT(*) OVER()`)
+#'   rather than a second scan.
 #' @export
 cog_balances <- function(govid = NULL, years, category = NULL,
                          per_capita = FALSE, adjust_to_year = NULL,
                          basis = c("harmonized", "raw"), recipe = NULL,
-                         state = NULL, type = NULL) {
+                         state = NULL, type = NULL,
+                         limit = NULL, offset = NULL) {
   call <- match.call()
   basis <- match.arg(basis, c("harmonized", "raw"))
   # Coerce FIRST, validate second: .validate_verb_inputs() asserts
@@ -91,6 +102,22 @@ cog_balances <- function(govid = NULL, years, category = NULL,
   # validator's own doc comment for the incident that made that matter.
   .validate_verb_inputs(govid, years, category, per_capita, adjust_to_year,
                         recipe)
+
+  # Same semantics as the money verbs (R/pagination.R). Only the `recipe`
+  # conflict applies here: cog_balances() has no `complete` argument, and a
+  # recipe's result comes from .run_recipe()'s own query, which pagination is
+  # not wired into.
+  paging <- .validate_pagination(limit, offset)
+  limit  <- paging$limit
+  offset <- paging$offset
+  if (!is.null(limit) && !is.null(recipe)) {
+    cli::cli_abort(c(
+      "`limit`/`offset` cannot be combined with `recipe`.",
+      "i" = "A recipe's result comes from a separate query (`.run_recipe()`) that pagination is not wired into yet.",
+      "*" = "Drop `limit`/`offset`, or drop `recipe`."
+    ), class = "uscogdata_recipe_pagination_conflict")
+  }
+
   years <- as.integer(years)
   if (!is.null(adjust_to_year)) adjust_to_year <- as.integer(adjust_to_year)
 
@@ -108,6 +135,7 @@ cog_balances <- function(govid = NULL, years, category = NULL,
   manifest <- .uscogdata_env$manifest
   recipe_block <- NULL
   category_for_prov <- category
+  total_rows <- NULL  # set below only when limit is non-NULL (non-recipe path)
 
   if (!is.null(recipe)) {
     .require_schema_v5(con, manifest, "recipe =")
@@ -125,8 +153,18 @@ cog_balances <- function(govid = NULL, years, category = NULL,
   } else {
     sql <- .build_verb_sql("balance_annotated", "balance_subtype",
                            cohort, years, category,
-                           ig_view = NULL, subtype_scope = NULL)
+                           ig_view = NULL, subtype_scope = NULL,
+                           limit = limit, offset = offset)
     result <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
+    if (!is.null(limit)) {
+      paged <- .take_pagination_total(result, con, function() {
+        .build_verb_sql("balance_annotated", "balance_subtype",
+                        cohort, years, category,
+                        ig_view = NULL, subtype_scope = NULL)
+      })
+      result     <- paged$result
+      total_rows <- paged$total_rows
+    }
   }
 
   # Order matters (matches .verb_spendrev()): per-capita first, so
@@ -158,6 +196,7 @@ cog_balances <- function(govid = NULL, years, category = NULL,
   .emit_balance_caveats(prov$balance_caveats)
 
   attr(result, "provenance") <- prov
+  if (!is.null(limit)) attr(result, "total_rows") <- total_rows
   result
 }
 

R/pagination.R

@@ -0,0 +1,81 @@
+# R/pagination.R
+#
+# Shared limit/offset machinery. #39 established the semantics inside
+# .verb_spendrev(); #57 extends them to cog_gov_search() and cog_balances(),
+# which is what made a single definition worth having: three inline copies of
+# "coerce, refuse, unwrap the count" would be three places for the meaning of
+# `total_rows` to drift.
+#
+# The SQL side stays in .build_verb_sql() (R/spending.R) -- it already wraps
+# the aggregate in an outer SELECT so COUNT(*) OVER() sees the post-GROUP-BY
+# row count rather than the pre-aggregation one, and that is the subtle part
+# worth not duplicating either.
+
+#' Coerce and check a limit/offset pair.
+#'
+#' Returns the coerced pair, or NULL for `limit` when no page was requested.
+#' `offset` defaults to 0 whenever `limit` is set, so a caller can supply just
+#' `limit` and get the first page.
+#'
+#' Conflicts with other arguments are deliberately NOT checked here: they
+#' differ per verb (`complete`/`recipe` for the money verbs, basket mode for
+#' `cog_gov_search()`, `recipe` alone for `cog_balances()`), and a shared
+#' function taking a list of conflict flags would be harder to read than the
+#' three explicit refusals at the call sites.
+#' @noRd
+.validate_pagination <- function(limit, offset) {
+  if (is.null(limit)) {
+    return(list(limit = NULL, offset = NULL))
+  }
+  limit <- as.integer(limit)
+  if (length(limit) != 1L || is.na(limit) || limit < 0L) {
+    cli::cli_abort("`limit` must be a single non-negative integer.",
+                   class = "uscogdata_invalid_pagination")
+  }
+  offset <- if (is.null(offset)) 0L else as.integer(offset)
+  if (length(offset) != 1L || is.na(offset) || offset < 0L) {
+    cli::cli_abort("`offset` must be a single non-negative integer.",
+                   class = "uscogdata_invalid_pagination")
+  }
+  list(limit = limit, offset = offset)
+}
+
+#' Wrap a query so one page comes back carrying the unpaginated total.
+#'
+#' `COUNT(*) OVER()` rides along as an ordinary column, so the caller gets the
+#' true total from the SAME scan instead of a second round trip. The outer
+#' `SELECT *` matters: appending LIMIT/OFFSET directly to a grouped query would
+#' have the window function count pre-aggregation rows.
+#' @noRd
+.paginate_sql <- function(base_sql, limit, offset) {
+  if (is.null(limit)) return(base_sql)
+  sprintf(
+    "SELECT *, COUNT(*) OVER() AS pagination_total_rows
+     FROM (%s) AS _paged
+     LIMIT %d OFFSET %d",
+    base_sql, limit, offset
+  )
+}
+
+#' Strip the count column back out and report the unpaginated total.
+#'
+#' Returns `list(result = , total_rows = )`.
+#'
+#' An empty page -- an offset past the end -- carries no row to read the window
+#' function off, so that one case falls back to a second, unpaginated
+#' `COUNT(*)` rather than reporting a wrong zero. `unpaged_sql` is passed as a
+#' function so the fallback query is only BUILT when it is actually needed;
+#' every caller's unpaginated SQL is otherwise constructed on every paged call
+#' and thrown away.
+#' @noRd
+.take_pagination_total <- function(result, con, unpaged_sql) {
+  if (nrow(result) > 0L) {
+    total <- result$pagination_total_rows[[1]]
+    result$pagination_total_rows <- NULL
+    return(list(result = result, total_rows = as.integer(total)))
+  }
+  count_sql <- sprintf("SELECT COUNT(*) AS n FROM (%s) AS _uncounted",
+                       if (is.function(unpaged_sql)) unpaged_sql() else unpaged_sql)
+  list(result = result,
+       total_rows = as.integer(DBI::dbGetQuery(con, count_sql)$n[[1]]))
+}

R/search.R

@@ -44,10 +44,22 @@
 #'   in basket mode (recycles from length 1). Excluded types `4`/`5` (or
 #'   `"special_district"` / `"school_district"`) trigger an explanatory
 #'   message and an empty result.
+#' @param limit Maximum number of rows to return, applied in SQL. `NULL`
+#'   (default) returns every match -- which, with no other filter, is the
+#'   entire crosswalk. Utility mode only: pagination has no meaning in basket
+#'   mode, where the result is one resolved row per requested name in input
+#'   order, and is refused there with class
+#'   `uscogdata_basket_pagination_conflict`.
+#' @param offset Rows to skip before `limit` starts counting (0-based).
+#'   Ignored if `limit` is `NULL`; defaults to `0L` when `limit` is set.
 #' @return A tibble of `canonical_fips_xwalk` rows. In utility mode, all
-#'   matches sorted by `population_acs` desc. In basket mode, resolved
-#'   rows in input order, with `attr(., "resolution")` set to the
-#'   sidecar tibble.
+#'   matches sorted by `population_acs` desc, ties broken by
+#'   `canonical_govid`. In basket mode, resolved rows in input order, with
+#'   `attr(., "resolution")` set to the sidecar tibble.
+#'
+#'   When `limit` is set, carries a `total_rows` attribute: the full
+#'   unpaginated match count, computed by the same query (`COUNT(*) OVER()`)
+#'   rather than a second scan.
 #' @seealso [cog_basket_resolution()], [cog_basket_unresolved()],
 #'   [cog_spending()], [cog_revenue()].
 #' @examples
@@ -82,7 +94,12 @@
 #' )
 #' }
 #' @export
-cog_gov_search <- function(name = NULL, state = NULL, type = NULL) {
+cog_gov_search <- function(name = NULL, state = NULL, type = NULL,
+                           limit = NULL, offset = NULL) {
+  paging <- .validate_pagination(limit, offset)
+  limit  <- paging$limit
+  offset <- paging$offset
+
   if (!is.null(type) && length(type) == 1L && .is_excluded_type(type)) {
     cli::cli_inform(c(
       i = "v0.1 covers gov_types 0-3 (state/county/city/township) only.",
@@ -94,6 +111,18 @@ cog_gov_search <- function(name = NULL, state = NULL, type = NULL) {
   con <- .ensure_session()
 
   if (length(name) > 1L) {
+    # Basket mode returns one resolved row per requested name, in input order,
+    # with a resolution sidecar describing how each was matched. A page of that
+    # is not a page of anything the caller asked for -- the sidecar would still
+    # describe every name -- so refuse rather than silently ignoring the
+    # arguments. Same shape as the recipe/complete refusals in .verb_spendrev().
+    if (!is.null(limit)) {
+      cli::cli_abort(c(
+        "`limit`/`offset` cannot be combined with basket mode.",
+        "i" = "Basket mode ({.code length(name) > 1}) returns one resolved row per requested name, in input order, with a resolution sidecar covering all of them.",
+        "*" = "Drop `limit`/`offset`, or search one name at a time."
+      ), class = "uscogdata_basket_pagination_conflict")
+    }
     return(.resolve_basket(name = name, state = state, type = type, con = con))
   }
 
@@ -123,12 +152,26 @@ cog_gov_search <- function(name = NULL, state = NULL, type = NULL) {
   }
 
   where <- if (length(preds) == 0L) "" else paste("WHERE", paste(preds, collapse = " AND "))
-  sql <- paste(
+  # canonical_govid breaks ties. population_acs alone is NOT a total order --
+  # governments sharing a population, and the whole NULLS LAST block, came back
+  # in whatever order the scan produced. That was invisible while every call
+  # returned the full result set, but it makes a paged sweep unsound: two
+  # requests can order the tied rows differently, so a row is duplicated on one
+  # page and missing from the next. Any pagination has to sit on a total order.
+  base_sql <- paste(
     "SELECT * FROM canonical_fips_xwalk",
     where,
-    "ORDER BY population_acs DESC NULLS LAST"
+    "ORDER BY population_acs DESC NULLS LAST, canonical_govid"
   )
-  tibble::as_tibble(DBI::dbGetQuery(con, sql))
+  result <- tibble::as_tibble(
+    DBI::dbGetQuery(con, .paginate_sql(base_sql, limit, offset))
+  )
+  if (is.null(limit)) return(result)
+
+  paged <- .take_pagination_total(result, con, base_sql)
+  out <- paged$result
+  attr(out, "total_rows") <- paged$total_rows
+  out
 }
 
 #' @noRd

R/spending.R

@@ -398,17 +398,10 @@ cog_spending <- function(govid = NULL, years, category = NULL,
   # up front rather than silently ignored: complete = TRUE fills a grid over
   # the FULL requested (year, category) space, and a recipe's result comes
   # from .run_recipe()'s own query, which this function does not touch.
+  paging <- .validate_pagination(limit, offset)
+  limit  <- paging$limit
+  offset <- paging$offset
   if (!is.null(limit)) {
-    limit <- as.integer(limit)
-    if (length(limit) != 1L || is.na(limit) || limit < 0L) {
-      cli::cli_abort("`limit` must be a single non-negative integer.",
-                     class = "uscogdata_invalid_pagination")
-    }
-    offset <- if (is.null(offset)) 0L else as.integer(offset)
-    if (length(offset) != 1L || is.na(offset) || offset < 0L) {
-      cli::cli_abort("`offset` must be a single non-negative integer.",
-                     class = "uscogdata_invalid_pagination")
-    }
     if (complete) {
       cli::cli_abort(c(
         "`limit`/`offset` cannot be combined with `complete = TRUE`.",
@@ -466,24 +459,14 @@ cog_spending <- function(govid = NULL, years, category = NULL,
                            limit = limit, offset = offset)
     result <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
     if (!is.null(limit)) {
-      # COUNT(*) OVER() rides along as an ordinary column so the total comes
-      # from the same scan when this page has any rows -- see
-      # .build_verb_sql(). An empty page (offset past the end) carries no
-      # such row to read it from, so that one case falls back to a second,
-      # unpaginated COUNT(*) query rather than reporting a wrong zero.
-      if (nrow(result) > 0L) {
-        total_rows <- result$pagination_total_rows[[1]]
-        result$pagination_total_rows <- NULL
-      } else {
-        count_sql <- sprintf(
-          "SELECT COUNT(*) AS n FROM (%s) AS _uncounted",
-          .build_verb_sql(view, subtype_col, cohort, years,
-                          if (all_categories) NULL else category,
-                          ig_view, subtype_scope,
-                          all_categories = all_categories)
-        )
-        total_rows <- as.integer(DBI::dbGetQuery(con, count_sql)$n[[1]])
-      }
+      paged <- .take_pagination_total(result, con, function() {
+        .build_verb_sql(view, subtype_col, cohort, years,
+                        if (all_categories) NULL else category,
+                        ig_view, subtype_scope,
+                        all_categories = all_categories)
+      })
+      result     <- paged$result
+      total_rows <- paged$total_rows
     }
   }
 
@@ -873,22 +856,9 @@ cog_spending <- function(govid = NULL, years, category = NULL,
   # matching row across the network only to slice and discard most of it
   # afterward (the pattern behind the 2026-08-06 production incident: a
   # 193,105-row/194-page sweep re-ran the full query and re-listified every
-  # row on EVERY page). COUNT(*) OVER() rides along as an ordinary column so
-  # the caller gets the true total from this same scan -- see the call site
-  # in .verb_spendrev(), which reads it off row 1 and strips it back out.
-  # The outer SELECT * wrapping (rather than appending LIMIT/OFFSET directly
-  # to base_sql) is what makes COUNT(*) OVER() see the post-GROUP-BY row
-  # count, not the pre-aggregation one.
-  if (is.null(limit)) {
-    base_sql
-  } else {
-    sprintf(
-      "SELECT *, COUNT(*) OVER() AS pagination_total_rows
-       FROM (%s) AS _paged
-       LIMIT %d OFFSET %d",
-      base_sql, limit, offset
-    )
-  }
+  # row on EVERY page). See .paginate_sql() in R/pagination.R for why the
+  # wrapping is an outer SELECT rather than a bare LIMIT on base_sql.
+  .paginate_sql(base_sql, limit, offset)
 }
 
 #' Join population onto a result and derive the per-capita columns.