Bound writer residency and seek wallet freshness by canonical order

Author: wock9000

docs/PUBLIC_API.md

@@ -263,9 +263,9 @@ responses distinguish complete collected-fee totals from partial evidence:
   `episode_selection: "last_activity_within_window"` and
   `financial_scope: "lifetime_of_selected_episodes"`. Do not treat these
   values as cash flows earned exclusively during the requested window.
-- Owner-list `financials.through_order` is the canonical
-  `[block_number, transaction_index, log_index]` boundary captured in the same
-  WAL snapshot as the financial aggregates. `financials.as_of` describes that
+- Owner-list `financials.through_order` contains the canonical `block_number`,
+  `tx_index`, and `log_index` boundary captured in the same WAL snapshot as the
+  financial aggregates. `financials.as_of` describes that
   boundary, not the time of the latest observed activity. Pending projection
   work or later activity keeps `financials.pending` true without advancing the
   captured boundary. A completed snapshot is distinct from complete history,

docs/RPC_OPERATIONS.md

@@ -70,6 +70,19 @@ SQLite's variable limit. This avoids handing the Python interpreter to competing
 valuation workers between every inserted row. The outer durable transaction
 preserves the raw event/cursor/job boundary.
 
+Adaptive scan sizing targets 50 ms of writer residency, half the observed
+100 ms block interval. Growth is capped by measured store throughput as well as
+log density, and history starts with eight blocks before adapting. This is a
+feedback target, not a hard transaction deadline; indivisible block work can
+exceed it. Multi-second history commits previously starved both the live lane
+and financial workers.
+
+Wallet freshness reads walk the canonical event-order index after checking
+whether the requested scope is empty. A timestamp-range plan sorted the entire
+recent ledger before returning its newest event and exceeded 30 seconds on the
+production snapshot; the ordered lookup returned the same boundary in under
+0.25 seconds including process startup.
+
 The CLI caps Python's thread-switch interval at 1 ms before starting workers,
 preserving an already-shorter interval. This reduces interpreter handoff delays
 when SQLite releases the GIL during ingestion while valuation threads run.

src/rhpools/lp_market_accounting.py

@@ -3690,13 +3690,20 @@ def _scoped_owner_financial_state(
             " WHERE " + " AND ".join(event_clauses)
             if event_clauses else ""
         )
-        through = conn.execute(
-            "SELECT e.block_number,e.tx_index,e.log_index,e.timestamp "
-            "FROM events e" + event_where
-            + " ORDER BY e.block_number DESC,e.tx_index DESC,"
-            "e.log_index DESC LIMIT 1",
-            event_args,
-        ).fetchone()
+        through = None
+        # A window lookup must not sort the entire recent ledger to find its
+        # newest event. Check empty scopes before walking the ordered index.
+        if not event_clauses or conn.execute(
+            "SELECT 1 FROM events e" + event_where + " LIMIT 1", event_args,
+        ).fetchone() is not None:
+            order_index = "" if pool_id else " INDEXED BY events_block_idx"
+            through = conn.execute(
+                "SELECT e.block_number,e.tx_index,e.log_index,e.timestamp "
+                "FROM events e" + order_index + event_where
+                + " ORDER BY e.block_number DESC,e.tx_index DESC,"
+                "e.log_index DESC LIMIT 1",
+                event_args,
+            ).fetchone()
         through_order = (
             {
                 "block_number": int(through["block_number"]),

src/rhpools/lp_market_index.py

@@ -59,9 +59,11 @@
 LIVE_MAX_CHUNK = 2_048
 RECENT_CATCHUP_PRIORITY_BLOCKS = 512
 HISTORY_MIN_CHUNK = 1
-HISTORY_INITIAL_CHUNK = 4_096
+HISTORY_INITIAL_CHUNK = 8
 HISTORY_MAX_CHUNK = 32_768
-MAX_INTERVAL_STORE_SECONDS = 2.0
+# Target half the chain's observed ~100 ms block interval, not multi-second
+# writer monopolies that stall live ingestion and every financial worker.
+MAX_INTERVAL_STORE_SECONDS = 0.05
 ENRICH_BATCH = 8
 ENRICHMENT_CAPABILITY_RECHECK_S = 300.0
 REPROJECT_BATCH = 128
@@ -3544,6 +3546,10 @@ def _resize_after_success(
                 current * 2,
                 sample_blocks * (MAX_LOGS_PER_RESPONSE * 4 // 5) // count,
             )
+        if store_seconds > 0:
+            candidate = min(candidate, max(
+                minimum, int(sample_blocks * MAX_INTERVAL_STORE_SECONDS / store_seconds),
+            ))
         if candidate <= current:
             return
         resized = min(maximum, max(current + 1, candidate))