Drop server item cap for bounded chunk queries

packages/cli/src/evm_hypersync_source/mod.rs

@@ -177,7 +177,7 @@ impl EvmHyperSyncClient {
                     .map(log_selection_from_built)
                     .collect(),
             ),
-            max_num_logs: Some(params.max_num_logs),
+            max_num_logs: params.max_num_logs,
             field_selection: query::FieldSelection {
                 block: Some(validated_block_fields.clone()),
                 transaction: Some(transaction_fields),
@@ -264,7 +264,8 @@ pub struct EventItemsQuery {
     pub from_block: i64,
     /// Inclusive; `None` queries to the end of available data.
     pub to_block: Option<i64>,
-    pub max_num_logs: i64,
+    /// `None` sends no server-side cap on the number of logs returned.
+    pub max_num_logs: Option<i64>,
     pub registration_indexes: Vec<i64>,
     pub addresses_by_contract_name: HashMap<String, Vec<String>>,
     /// Contract names to fetch address-free even though their registrations

packages/cli/src/svm_hypersync_source/mod.rs

@@ -296,7 +296,9 @@ impl SvmHyperSyncClient {
                 .map_err(map_err)?,
             instructions: built.instruction_selections.clone(),
             field_selection,
-            max_num_instructions: usize::try_from(params.max_num_instructions).ok(),
+            max_num_instructions: params
+                .max_num_instructions
+                .and_then(|v| usize::try_from(v).ok()),
             ..Default::default()
         };
 
@@ -350,7 +352,8 @@ pub struct EventItemsQuery {
     pub from_slot: i64,
     /// Inclusive; `None` queries to the end of available data.
     pub to_slot: Option<i64>,
-    pub max_num_instructions: i64,
+    /// `None` sends no server-side cap on the number of instructions returned.
+    pub max_num_instructions: Option<i64>,
     pub registration_indexes: Vec<i64>,
     pub addresses_by_contract_name: HashMap<String, Vec<String>>,
 }

packages/envio-tests/test/HyperSync_test.res

@@ -38,7 +38,7 @@ describe_skip("Test Hyperliquid broken transaction response", () => {
       ),
       ~fromBlock=12403138,
       ~toBlock=Some(12403139),
-      ~maxNumLogs=5000,
+      ~maxNumLogs=Some(5000),
       ~registrationIndexes=[0],
       ~addressesByContractName=Dict.make(),
       ~clientFilteredContracts=None,

packages/envio-tests/test/SvmHyperSyncSource_test.res

@@ -167,7 +167,7 @@ describe("SvmHyperSyncSource.getItemsOrThrow (mocked client)", () => {
         ~contractNameByAddress,
         ~knownHeight=slot + 1000,
         ~partitionId="0",
-        ~itemsTarget=5000,
+        ~itemsTarget=Some(5000),
         ~selection={
           onEventRegistrations: [reg],
           dependsOnAddresses: true,

packages/envio/src/ChainState.res

@@ -547,13 +547,7 @@ let frontierProgress = (cs: t) =>
 // maxTargetBlock set to the most-behind chain's progress mapped onto this
 // chain, so a chain with budget can't run further ahead than the chain the
 // whole pool is prioritizing.
-let getNextQuery = (
-  cs: t,
-  ~chainTargetItems: float,
-  ~chunkItemsMultiplier=1.,
-  ~itemsTargetFloor=0,
-  ~maxTargetBlock=?,
-) => {
+let getNextQuery = (cs: t, ~chainTargetItems: float, ~maxTargetBlock=?) => {
   let chainTargetBlock = cs->targetBlock(~chainTargetItems)
   let chainTargetBlock = switch maxTargetBlock {
   | Some(maxTargetBlock) => Pervasives.min(chainTargetBlock, maxTargetBlock)
@@ -567,23 +561,14 @@ let getNextQuery = (
   // of being held by an oversized probe.
   let chainTargetItems = switch cs->effectiveDensity {
   | Some(density) if density > 0. =>
-    // No extra headroom here: the budget is reserved in honest itemsEst units,
-    // and truncation safety lives in the itemsTarget server cap (sized with
-    // chunkItemsMultiplier at query creation) — multiplying the budget cap too
-    // would compound the two and hold budget away from other chains.
     let rangeCost =
       density *. (chainTargetBlock - cs.fetchState->FetchState.bufferBlockNumber)->Int.toFloat
     Pervasives.min(chainTargetItems, Math.ceil(rangeCost) +. cs.pendingBudget)
   // No density signal: the cross-chain waterfall already clamped the handed
   // budget to the cold-chain cap, so it's used as-is.
   | _ => chainTargetItems
   }
-  cs.fetchState->FetchState.getNextQuery(
-    ~chainTargetBlock,
-    ~chainTargetItems,
-    ~chunkItemsMultiplier,
-    ~itemsTargetFloor,
-  )
+  cs.fetchState->FetchState.getNextQuery(~chainTargetBlock, ~chainTargetItems)
 }
 
 // Run a fetch tick for this chain against its sources, feeding the owned fetch

packages/envio/src/ChainState.resi

@@ -80,13 +80,7 @@ let hasReadyItem: t => bool
 let targetBlock: (t, ~chainTargetItems: float) => int
 let blockAtProgress: (t, ~progress: float) => int
 let frontierProgress: t => float
-let getNextQuery: (
-  t,
-  ~chainTargetItems: float,
-  ~chunkItemsMultiplier: float=?,
-  ~itemsTargetFloor: int=?,
-  ~maxTargetBlock: int=?,
-) => FetchState.nextQuery
+let getNextQuery: (t, ~chainTargetItems: float, ~maxTargetBlock: int=?) => FetchState.nextQuery
 let dispatch: (
   t,
   ~executeQuery: FetchState.query => promise<unit>,

packages/envio/src/CrossChainState.res

@@ -249,19 +249,6 @@ let checkAndFetch = async (
   // while it takes its first measurements. Its probe is one admission unit.
   let coldChainBudget = minimumAdmissionBudget
 
-  // Chunk reservations get headroom over the density estimate so a
-  // denser-than-expected range doesn't truncate at the server cap; realtime
-  // gets more since a forced catch-up query there costs a head-poll roundtrip.
-  let chunkItemsMultiplier = crossChainState.isRealtime ? 3. : 1.5
-
-  // Server-cap floor for bounded queries: their block range is already the
-  // hard bound on the response, so a low density estimate shrinking the cap
-  // below this only buys self-truncated responses. Splitting the target pool
-  // across a chain's concurrency slots keeps the worst case — every in-flight
-  // bounded query returning a full floored response at once — at ~one buffer
-  // target.
-  let itemsTargetFloor = crossChainState.targetBufferSize / FetchState.maxChainConcurrency
-
   let prioritizedChainStates = crossChainState->priorityOrder
 
   // Alignment anchor: the first known-height chain in priority order — which,
@@ -308,12 +295,7 @@ let checkAndFetch = async (
         Some(cs->ChainState.blockAtProgress(~progress=progress +. 0.1))
       | _ => None
       }
-      switch cs->ChainState.getNextQuery(
-        ~chainTargetItems,
-        ~chunkItemsMultiplier,
-        ~itemsTargetFloor,
-        ~maxTargetBlock?,
-      ) {
+      switch cs->ChainState.getNextQuery(~chainTargetItems, ~maxTargetBlock?) {
       | WaitingForNewBlock as action => actionByChain->Utils.Dict.setByInt(chainId, action)
       | NothingToQuery =>
         // A chain below its head can emit no query when its budget went to
@@ -333,7 +315,7 @@ let checkAndFetch = async (
               {
                 "fromBlock": query.fromBlock,
                 "targetBlock": query.toBlock,
-                "targetEvents": query.itemsTarget,
+                "targetEvents": query.itemsEst,
               },
             )
           )