Lift the transfer filter and page cut above the Store seam

query_transfers and query_postings had copied the same filter closure and the
same total/skip/take pagination tail on both sides of the Store trait.  Worse,
the two query_transfers adapters answered an account-less query differently
behind one signature: the in-memory default returned an error while SQL ran a
store-wide scan, a divergence no conformance test held them to. The dashboard
already relied on the scan (it queries with a default, account-less filter), so
the in-memory backend would have failed there.

Introduce a query module in kuatia-storage that states the contract once:
filter_transfers for the time-window and book predicates, and paginate for the
total plus offset/limit cut. Each backend now only loads its candidate records
and hands them to the shared code; SQL keeps its genuine LIMIT push-down for
postings. query_transfers loses its trait default so both backends must
implement the load, and both now scan store-wide when no account is given. A
new conformance test pins that shared answer.

crates/kuatia-storage-sql/src/lib.rs

@@ -19,6 +19,7 @@ use sqlx::{Any, Pool, Row};
 
 use kuatia_storage::error::StoreError;
 use kuatia_storage::events::{EventStore, LedgerEvent};
+use kuatia_storage::query::{filter_transfers, paginate};
 use kuatia_storage::store::*;
 use kuatia_types::*;
 
@@ -1194,35 +1195,14 @@ impl TransferStore for SqlStore {
             records
         };
 
-        // Filter in memory for remaining conditions.
-        let filtered: Vec<EnvelopeRecord> = base_records
-            .into_iter()
-            .filter(|r| {
-                if let Some(from) = query.from_ts
-                    && r.created_at < from
-                {
-                    return false;
-                }
-                if let Some(to) = query.to_ts
-                    && r.created_at >= to
-                {
-                    return false;
-                }
-                if let Some(book) = query.book
-                    && r.envelope.book() != book
-                {
-                    return false;
-                }
-                true
-            })
-            .collect();
-
-        let total = filtered.len() as u64;
-        let offset = query.offset.unwrap_or(0) as usize;
-        let limit = query.limit.unwrap_or(u32::MAX) as usize;
-        let items = filtered.into_iter().skip(offset).take(limit).collect();
-
-        Ok(Page { items, total })
+        // The account/subaccount narrowing happened in the load above; the
+        // shared filter covers the time-window and book predicates, then the
+        // shared page cut applies `offset`/`limit`.
+        Ok(paginate(
+            filter_transfers(base_records, query),
+            query.offset,
+            query.limit,
+        ))
     }
 }
 

crates/kuatia-storage/src/lib.rs

@@ -6,5 +6,6 @@
 pub mod error;
 pub mod events;
 pub mod mem_store;
+pub mod query;
 pub mod store;
 pub mod store_tests;

crates/kuatia-storage/src/mem_store.rs

@@ -14,8 +14,10 @@ use kuatia_types::{
 
 use crate::error::StoreError;
 use crate::events::{EventStore, LedgerEvent};
+use crate::query::{filter_transfers, paginate};
 use crate::store::{
-    AccountStore, BookStore, EnvelopeRecord, PostingStore, SagaStore, TransferStore,
+    AccountStore, BookStore, EnvelopeRecord, Page, PostingStore, SagaStore, TransferQuery,
+    TransferStore,
 };
 
 /// Postings held as an immutable record table plus two index maps that carry
@@ -355,6 +357,30 @@ impl TransferStore for InMemoryStore {
         result.sort_by_key(|r| r.created_at);
         Ok(result)
     }
+
+    async fn query_transfers(
+        &self,
+        query: &TransferQuery,
+    ) -> Result<Page<EnvelopeRecord>, StoreError> {
+        // Load candidates: narrow by the participation index when an account is
+        // given, otherwise scan every stored transfer (the same store-wide
+        // answer the SQL backend gives for `account == None`). Then apply the
+        // shared time-window/book filter and page cut.
+        let candidates = match query.account {
+            Some(account) => self.get_transfers_for_account(account, query.sub).await?,
+            None => {
+                let transfers = self.transfers.read().await;
+                let mut all: Vec<EnvelopeRecord> = transfers.values().cloned().collect();
+                all.sort_by_key(|r| r.created_at);
+                all
+            }
+        };
+        Ok(paginate(
+            filter_transfers(candidates, query),
+            query.offset,
+            query.limit,
+        ))
+    }
 }
 
 // ---------------------------------------------------------------------------

crates/kuatia-storage/src/query.rs

@@ -0,0 +1,59 @@
+//! Filter and pagination primitives shared by every [`Store`](crate::store::Store)
+//! backend, stated once above the trait seam.
+//!
+//! A backend's `query_transfers`/`query_postings` implementation does only what
+//! is genuinely backend-specific: load the candidate records (via an account
+//! index, a store-wide scan, or a SQL `LIMIT` push-down). Everything after that,
+//! the time-window/book predicate and the `total` + `skip`/`take` cut, is the
+//! same contract regardless of backend and lives here.
+
+use crate::store::{EnvelopeRecord, Page, TransferQuery};
+
+/// Keep only the transfers matching a query's time-window and book predicates.
+///
+/// The account/subaccount filter is *not* applied here: a backend narrows to
+/// participating accounts when it loads candidates (an in-memory participation
+/// index or the SQL `transfer_accounts` join), because that filter is what the
+/// backend can push down. This covers every remaining predicate so both
+/// backends agree on the contract.
+pub fn filter_transfers(
+    records: Vec<EnvelopeRecord>,
+    query: &TransferQuery,
+) -> Vec<EnvelopeRecord> {
+    records
+        .into_iter()
+        .filter(|r| {
+            if let Some(from) = query.from_ts
+                && r.created_at < from
+            {
+                return false;
+            }
+            if let Some(to) = query.to_ts
+                && r.created_at >= to
+            {
+                return false;
+            }
+            if let Some(book) = query.book
+                && r.envelope.book() != book
+            {
+                return false;
+            }
+            true
+        })
+        .collect()
+}
+
+/// Cut a fully-filtered, ordered record set into one page: `total` is the
+/// pre-pagination count, then skip `offset` (default 0) and take `limit`
+/// (default unbounded).
+///
+/// Callers that push `LIMIT`/`OFFSET` into the store (e.g. SQL `query_postings`)
+/// build their own [`Page`] and skip this; it exists for the backends that hold
+/// the full candidate set in memory.
+pub fn paginate<T>(records: Vec<T>, offset: Option<u32>, limit: Option<u32>) -> Page<T> {
+    let total = records.len() as u64;
+    let offset = offset.unwrap_or(0) as usize;
+    let limit = limit.unwrap_or(u32::MAX) as usize;
+    let items = records.into_iter().skip(offset).take(limit).collect();
+    Page { items, total }
+}

crates/kuatia-storage/src/store.rs

@@ -170,15 +170,12 @@ pub trait PostingStore: Send + Sync {
     /// Query postings with filtering and pagination.
     async fn query_postings(&self, query: &PostingQuery) -> Result<Page<Posting>, StoreError> {
         // `get_postings_by_account` returns a deterministic id-ordered sequence,
-        // so LIMIT/OFFSET here paginate stably without an extra sort.
+        // so paginating here is stable without an extra sort. A backend that can
+        // push the window into its query (e.g. SQL `LIMIT`) overrides this.
         let all = self
             .get_postings_by_account(query.account, query.sub, query.asset.as_ref(), query.filter)
             .await?;
-        let total = all.len() as u64;
-        let offset = query.offset.unwrap_or(0) as usize;
-        let limit = query.limit.unwrap_or(u32::MAX) as usize;
-        let items = all.into_iter().skip(offset).take(limit).collect();
-        Ok(Page { items, total })
+        Ok(crate::query::paginate(all, query.offset, query.limit))
     }
 }
 
@@ -209,48 +206,18 @@ pub trait TransferStore: Send + Sync {
     ) -> Result<Vec<EnvelopeRecord>, StoreError>;
 
     /// Query transfers with filtering and pagination.
+    ///
+    /// A backend loads the candidate records, then hands them to
+    /// [`filter_transfers`](crate::query::filter_transfers) and
+    /// [`paginate`](crate::query::paginate) for the shared time-window/book and
+    /// page cut. There is no default because loading candidates differs by
+    /// backend and, crucially, an `account == None` query must be answered the
+    /// same way everywhere: a store-wide scan, not an error. Every backend does
+    /// exactly that.
     async fn query_transfers(
         &self,
         query: &TransferQuery,
-    ) -> Result<Page<EnvelopeRecord>, StoreError> {
-        // Default in-memory implementation
-        let all = if let Some(account) = query.account {
-            self.get_transfers_for_account(account, query.sub).await?
-        } else {
-            return Err(StoreError::Internal(
-                "query_transfers requires account filter in default implementation".into(),
-            ));
-        };
-
-        let filtered: Vec<EnvelopeRecord> = all
-            .into_iter()
-            .filter(|r| {
-                if let Some(from) = query.from_ts
-                    && r.created_at < from
-                {
-                    return false;
-                }
-                if let Some(to) = query.to_ts
-                    && r.created_at >= to
-                {
-                    return false;
-                }
-                if let Some(book) = query.book
-                    && r.envelope.book() != book
-                {
-                    return false;
-                }
-                true
-            })
-            .collect();
-
-        let total = filtered.len() as u64;
-        let offset = query.offset.unwrap_or(0) as usize;
-        let limit = query.limit.unwrap_or(u32::MAX) as usize;
-        let items = filtered.into_iter().skip(offset).take(limit).collect();
-
-        Ok(Page { items, total })
-    }
+    ) -> Result<Page<EnvelopeRecord>, StoreError>;
 }
 
 /// Saga state persistence for crash recovery.

crates/kuatia-storage/src/store_tests.rs

@@ -1052,6 +1052,36 @@ pub async fn query_transfers_by_book(store: &(impl Store + 'static)) {
     assert_eq!(page.items[0].envelope.book(), BookId(5));
 }
 
+/// An `account == None` query is a store-wide scan on every backend: it returns
+/// all transfers regardless of participation, and the time-window/book filters
+/// still apply. This pins the account-optional contract that used to diverge
+/// (in-memory errored, SQL scanned).
+pub async fn query_transfers_store_wide(store: &(impl Store + 'static)) {
+    let (e1, t1) = make_envelope_with_book(BookId(1));
+    commit_envelope(store, e1, t1, 1000).await;
+
+    let (e2, t2) = make_envelope_with_book(BookId(2));
+    commit_envelope(store, e2, t2, 2000).await;
+
+    // No account filter: both transfers come back, newest bound honored.
+    let page = store
+        .query_transfers(&TransferQuery::default())
+        .await
+        .unwrap();
+    assert_eq!(page.total, 2);
+
+    // The shared book filter still narrows a store-wide scan.
+    let page = store
+        .query_transfers(&TransferQuery {
+            book: Some(BookId(2)),
+            ..Default::default()
+        })
+        .await
+        .unwrap();
+    assert_eq!(page.total, 1);
+    assert_eq!(page.items[0].created_at, 2000);
+}
+
 // ---------------------------------------------------------------------------
 // SagaStore tests
 // ---------------------------------------------------------------------------
@@ -1235,6 +1265,7 @@ macro_rules! store_tests {
             query_transfers_by_date_range,
             query_transfers_pagination,
             query_transfers_by_book,
+            query_transfers_store_wide,
             // SagaStore
             save_and_list_sagas,
             delete_saga,