From eac19466a5051f8258040a3783795b4cfea79140 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 24 Jul 2026 15:31:32 +0000 Subject: [PATCH 1/2] Drop server item cap for bounded chunk queries A chunk query with a specific end block already has its toBlock as a hard bound on the response range, so the itemsTarget/maxNumLogs cap only exists to guard against a denser-than-expected range. With client-side address filtering the server counts filtered-out items toward that cap, so it can truncate the range short and open a gap that then needs a gap-fill roundtrip. Stop sending a server cap for bounded chunk queries: SourceManager forwards None to the source when the query is a chunk with a set toBlock, and the maxNumLogs / maxNumInstructions inputs become optional through the HyperSync clients and the napi boundary. A bounded chunk's partial response is now always treated as genuine source-capacity evidence. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01GqyCTGE239bpYGZ9UvfJDn --- packages/cli/src/evm_hypersync_source/mod.rs | 5 ++- packages/cli/src/svm_hypersync_source/mod.rs | 7 +++- packages/envio-tests/test/HyperSync_test.res | 2 +- .../test/SvmHyperSyncSource_test.res | 2 +- packages/envio/src/FetchState.res | 4 +- packages/envio/src/sources/HyperSync.res | 2 +- packages/envio/src/sources/HyperSync.resi | 2 +- .../envio/src/sources/HyperSyncClient.res | 3 +- packages/envio/src/sources/Source.res | 6 ++- packages/envio/src/sources/SourceManager.res | 6 ++- .../envio/src/sources/SvmHyperSyncClient.res | 3 +- .../envio/src/sources/SvmHyperSyncSource.res | 2 +- .../test/RpcSourceContract_test.res | 8 ++-- .../test_codegen/test/RpcSource_test.res | 18 ++++----- .../test/SourceBlockHashes_test.res | 2 +- .../test/lib_tests/FetchState_test.res | 38 +++++++++++++------ 16 files changed, 69 insertions(+), 41 deletions(-) diff --git a/packages/cli/src/evm_hypersync_source/mod.rs b/packages/cli/src/evm_hypersync_source/mod.rs index e574936ce..9b3a6979d 100644 --- a/packages/cli/src/evm_hypersync_source/mod.rs +++ b/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, - pub max_num_logs: i64, + /// `None` sends no server-side cap on the number of logs returned. + pub max_num_logs: Option, pub registration_indexes: Vec, pub addresses_by_contract_name: HashMap>, /// Contract names to fetch address-free even though their registrations diff --git a/packages/cli/src/svm_hypersync_source/mod.rs b/packages/cli/src/svm_hypersync_source/mod.rs index 86ef3df8a..3e002bf22 100644 --- a/packages/cli/src/svm_hypersync_source/mod.rs +++ b/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, - pub max_num_instructions: i64, + /// `None` sends no server-side cap on the number of instructions returned. + pub max_num_instructions: Option, pub registration_indexes: Vec, pub addresses_by_contract_name: HashMap>, } diff --git a/packages/envio-tests/test/HyperSync_test.res b/packages/envio-tests/test/HyperSync_test.res index 97fcac576..9702541ae 100644 --- a/packages/envio-tests/test/HyperSync_test.res +++ b/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, diff --git a/packages/envio-tests/test/SvmHyperSyncSource_test.res b/packages/envio-tests/test/SvmHyperSyncSource_test.res index 547a97d8c..2bed29266 100644 --- a/packages/envio-tests/test/SvmHyperSyncSource_test.res +++ b/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, diff --git a/packages/envio/src/FetchState.res b/packages/envio/src/FetchState.res index 4060d3135..e74ca1eb5 100644 --- a/packages/envio/src/FetchState.res +++ b/packages/envio/src/FetchState.res @@ -549,7 +549,9 @@ module OptimizedPartitions = { // Partial response is direct capacity evidence — unless it was // truncated by our own itemsTarget cap: that reflects the // reservation we asked for, not what the server could return. - itemsCount < query.itemsTarget + // Bounded chunks send no cap (see SourceManager), so their partial + // responses are always genuine source-capacity evidence. + query.isChunk || itemsCount < query.itemsTarget } else { // A full response updates only when the query's intended range // covers at least the partition's current chunk range — meaning it diff --git a/packages/envio/src/sources/HyperSync.res b/packages/envio/src/sources/HyperSync.res index 8aab3bcec..a05b089ac 100644 --- a/packages/envio/src/sources/HyperSync.res +++ b/packages/envio/src/sources/HyperSync.res @@ -98,7 +98,7 @@ module GetLogs = { let query: HyperSyncClient.EventItems.query = { fromBlock, toBlock, - maxNumLogs, + ?maxNumLogs, registrationIndexes, addressesByContractName, clientFilteredContracts, diff --git a/packages/envio/src/sources/HyperSync.resi b/packages/envio/src/sources/HyperSync.resi index 96a0b95be..e720b43c7 100644 --- a/packages/envio/src/sources/HyperSync.resi +++ b/packages/envio/src/sources/HyperSync.resi @@ -32,7 +32,7 @@ module GetLogs: { ~client: HyperSyncClient.t, ~fromBlock: int, ~toBlock: option, - ~maxNumLogs: int, + ~maxNumLogs: option, ~registrationIndexes: array, ~addressesByContractName: dict>, ~clientFilteredContracts: option>, diff --git a/packages/envio/src/sources/HyperSyncClient.res b/packages/envio/src/sources/HyperSyncClient.res index 13256f40c..3565e8c41 100644 --- a/packages/envio/src/sources/HyperSyncClient.res +++ b/packages/envio/src/sources/HyperSyncClient.res @@ -309,7 +309,8 @@ module EventItems = { fromBlock: int, // Inclusive; None queries to the end of available data. toBlock: option, - maxNumLogs: int, + // Absent means no server-side cap on the number of logs returned. + maxNumLogs?: int, registrationIndexes: array, addressesByContractName: dict>, // Contract names to fetch address-free even though their registrations diff --git a/packages/envio/src/sources/Source.res b/packages/envio/src/sources/Source.res index cd352e93b..98c903a52 100644 --- a/packages/envio/src/sources/Source.res +++ b/packages/envio/src/sources/Source.res @@ -82,8 +82,10 @@ type t = { // source should ask its backend for, from the query's own estResponseSize. // A HyperSync-backed source enforces it server-side, so a wrong estimate // truncates the response instead of overshooting the shared buffer. Sources - // without an equivalent lever (RPC, Fuel, Simulate) ignore it. - ~itemsTarget: int, + // without an equivalent lever (RPC, Fuel, Simulate) ignore it. None means no + // cap: bounded chunk queries fetch their whole range even if denser than + // expected, so client-side-filtered items can't truncate the range short. + ~itemsTarget: option, ~retry: int, ~logger: Pino.t, ) => promise, diff --git a/packages/envio/src/sources/SourceManager.res b/packages/envio/src/sources/SourceManager.res index f1123dfa0..6f0bacd67 100644 --- a/packages/envio/src/sources/SourceManager.res +++ b/packages/envio/src/sources/SourceManager.res @@ -768,7 +768,11 @@ let executeQuery = async ( ~partitionId=query.partitionId, ~knownHeight, ~selection=query.selection, - ~itemsTarget=query.itemsTarget, + // Bounded chunk queries send no server cap: their toBlock is already the + // hard bound, and with client-side address filtering the server counts + // filtered-out items toward the cap, truncating the range short and + // opening gaps. Everything else keeps its cap. + ~itemsTarget=query.isChunk && query.toBlock->Option.isSome ? None : Some(query.itemsTarget), ~retry, ~logger, ) diff --git a/packages/envio/src/sources/SvmHyperSyncClient.res b/packages/envio/src/sources/SvmHyperSyncClient.res index 598dcfbbe..81670b707 100644 --- a/packages/envio/src/sources/SvmHyperSyncClient.res +++ b/packages/envio/src/sources/SvmHyperSyncClient.res @@ -188,7 +188,8 @@ module EventItems = { fromSlot: int, // Inclusive; None queries to the end of available data. toSlot: option, - maxNumInstructions: int, + // Absent means no server-side cap on the number of instructions returned. + maxNumInstructions?: int, registrationIndexes: array, addressesByContractName: dict>, } diff --git a/packages/envio/src/sources/SvmHyperSyncSource.res b/packages/envio/src/sources/SvmHyperSyncSource.res index 55315746f..61557de04 100644 --- a/packages/envio/src/sources/SvmHyperSyncSource.res +++ b/packages/envio/src/sources/SvmHyperSyncSource.res @@ -100,7 +100,7 @@ let make = ( let query: SvmHyperSyncClient.EventItems.query = { fromSlot: fromBlock, toSlot: toBlock, - maxNumInstructions: itemsTarget, + maxNumInstructions: ?itemsTarget, registrationIndexes: selection.onEventRegistrations->Array.map(reg => reg.index), addressesByContractName, } diff --git a/scenarios/test_codegen/test/RpcSourceContract_test.res b/scenarios/test_codegen/test/RpcSourceContract_test.res index 6dab69050..6b09cd97a 100644 --- a/scenarios/test_codegen/test/RpcSourceContract_test.res +++ b/scenarios/test_codegen/test/RpcSourceContract_test.res @@ -95,7 +95,7 @@ let invoke = (source: Source.t, ~registration: Internal.evmOnEventRegistration, dependsOnAddresses: true, onEventRegistrations: [(registration :> Internal.onEventRegistration)], }, - ~itemsTarget=5_000, + ~itemsTarget=Some(5_000), ~retry, ~logger=Logging.createChild(~params={"test": "RPC source contract pin"}), ) @@ -444,7 +444,7 @@ let registerContractTests = (~name, ~factory: sourceFactory) => { dependsOnAddresses: true, onEventRegistrations: [(registration :> Internal.onEventRegistration)], }, - ~itemsTarget=5_000, + ~itemsTarget=Some(5_000), ~retry=0, ~logger=Logging.createChild(~params={"test": "RPC interval pin"}), ) @@ -585,7 +585,7 @@ let registerContractTests = (~name, ~factory: sourceFactory) => { dependsOnAddresses: false, onEventRegistrations: [(registration :> Internal.onEventRegistration)], }, - ~itemsTarget=5_000, + ~itemsTarget=Some(5_000), ~retry=0, ~logger=Logging.createChild(~params={"test": "RPC skip-all pin"}), ) @@ -707,7 +707,7 @@ let registerContractTests = (~name, ~factory: sourceFactory) => { (eventB :> Internal.onEventRegistration), ], }, - ~itemsTarget=5_000, + ~itemsTarget=Some(5_000), ~retry=0, ~logger=Logging.createChild(~params={"test": "RPC contract scoping pin"}), ) diff --git a/scenarios/test_codegen/test/RpcSource_test.res b/scenarios/test_codegen/test/RpcSource_test.res index 7e7a010c7..61d35af7c 100644 --- a/scenarios/test_codegen/test/RpcSource_test.res +++ b/scenarios/test_codegen/test/RpcSource_test.res @@ -748,7 +748,7 @@ describe("RpcSource - empty selection", () => { ~knownHeight=1, ~partitionId="0", ~selection={dependsOnAddresses: true, onEventRegistrations: []}, - ~itemsTarget=5000, + ~itemsTarget=Some(5000), ~retry=0, ~logger=Logging.createChild(~params={"test": "RpcSource empty selection"}), ) @@ -838,7 +838,7 @@ describe("RpcSource - getItemsOrThrow on response-too-large", () => { dependsOnAddresses: true, onEventRegistrations: [(eventConfig :> Internal.onEventRegistration)], }, - ~itemsTarget=5000, + ~itemsTarget=Some(5000), ~retry=0, ~logger=Logging.createChild(~params={"test": "RpcSource response too large"}), ) @@ -967,7 +967,7 @@ describe("RpcSource - getItemsOrThrow on response-too-large", () => { dependsOnAddresses: true, onEventRegistrations: [(eventConfig :> Internal.onEventRegistration)], }, - ~itemsTarget=5000, + ~itemsTarget=Some(5000), ~retry=0, ~logger=Logging.createChild(~params={"test": "RpcSource re-grow"}), ) @@ -1116,7 +1116,7 @@ describe("RpcSource - getItemsOrThrow classifies real provider block-range error dependsOnAddresses: true, onEventRegistrations: [(eventConfig :> Internal.onEventRegistration)], }, - ~itemsTarget=5000, + ~itemsTarget=Some(5000), ~retry=0, ~logger=Logging.createChild(~params={"test": "RpcSource classify " ++ name}), ) @@ -1211,7 +1211,7 @@ describe("RpcSource - getItemsOrThrow with missing transaction data", () => { dependsOnAddresses: true, onEventRegistrations: [(eventConfig :> Internal.onEventRegistration)], }, - ~itemsTarget=5000, + ~itemsTarget=Some(5000), ~retry, ~logger=Logging.createChild(~params={"test": "RpcSource missing transaction data"}), ) @@ -1361,7 +1361,7 @@ describe("RpcSource - getItemsOrThrow fans out multiple selections", () => { dependsOnAddresses: true, onEventRegistrations: [(eventConfig :> Internal.onEventRegistration)], }, - ~itemsTarget=5000, + ~itemsTarget=Some(5000), ~retry=0, ~logger=Logging.createChild(~params={"test": "RpcSource fan-out"}), ) @@ -1489,7 +1489,7 @@ describe("RpcSource - builds partition log selections end to end", () => { (reg :> Internal.onEventRegistration) ), }, - ~itemsTarget=5000, + ~itemsTarget=Some(5000), ~retry=0, ~logger=Logging.createChild(~params={"test": "RpcSource selection e2e"}), ) @@ -1577,7 +1577,7 @@ describe("RpcSource - getItemsOrThrow with a skip-all event filter", () => { dependsOnAddresses: false, onEventRegistrations: [(eventConfig :> Internal.onEventRegistration)], }, - ~itemsTarget=5000, + ~itemsTarget=Some(5000), ~retry=0, ~logger=Logging.createChild(~params={"test": "RpcSource skip-all"}), ) @@ -1738,7 +1738,7 @@ describe("RpcSource - getItemsOrThrow scopes filters to each contract's addresse dependsOnAddresses: true, onEventRegistrations: [(eventA :> Internal.onEventRegistration), (eventB :> Internal.onEventRegistration)], }, - ~itemsTarget=5000, + ~itemsTarget=Some(5000), ~retry=0, ~logger=Logging.createChild(~params={"test": "RpcSource pooled leak"}), ) diff --git a/scenarios/test_codegen/test/SourceBlockHashes_test.res b/scenarios/test_codegen/test/SourceBlockHashes_test.res index 0d17caf56..c0835ea4d 100644 --- a/scenarios/test_codegen/test/SourceBlockHashes_test.res +++ b/scenarios/test_codegen/test/SourceBlockHashes_test.res @@ -122,7 +122,7 @@ let invoke = async (source: Source.t, ~fromBlock, ~toBlock) => { ~knownHeight=toBlock + 1000, ~partitionId="0", ~selection=makeSelection(), - ~itemsTarget=5000, + ~itemsTarget=Some(5000), ~retry=0, ~logger=Logging.createChild(~params={"test": "SourceBlockHashes"}), ) catch { diff --git a/scenarios/test_codegen/test/lib_tests/FetchState_test.res b/scenarios/test_codegen/test/lib_tests/FetchState_test.res index 218a76a03..d1426012e 100644 --- a/scenarios/test_codegen/test/lib_tests/FetchState_test.res +++ b/scenarios/test_codegen/test/lib_tests/FetchState_test.res @@ -4724,11 +4724,12 @@ describe("FetchState.getNextQuery chunk headroom and budget-driven emit", () => }) describe("Response density and source range capacity update independently", () => { - // Source range capacity 300 with a pending chunk truncated at block 90: when - // the truncation was caused by our own itemsTarget cap it says nothing about - // server capacity, so the 300 history must survive. Its items/block ratio is - // still current density evidence and is blended 1:1 with the stored density. - // A sub-cap partial updates both signals. + // Source range capacity 300 with a bounded query truncated at block 90: for a + // non-chunk query the truncation may be our own itemsTarget cap, which says + // nothing about server capacity, so the 300 history must survive. Bounded + // chunks send no cap, so their partial responses are always genuine capacity + // evidence. Either way the items/block ratio is current density evidence, + // blended 1:1 with the stored density. let normalSelection = {FetchState.dependsOnAddresses: false, onEventRegistrations: []} let addressesByContractName = Dict.fromArray([("MockContract", [mockAddress0])]) @@ -4769,23 +4770,25 @@ describe("Response density and source range capacity update independently", () = clientFilterAddressThreshold: None, } - let chunkQuery: FetchState.query = { + let makeQuery = (~isChunk): FetchState.query => { partitionId: "0", fromBlock: 1, toBlock: Some(540), - isChunk: true, + isChunk, itemsTarget: 3, itemsEst: 3, selection: normalSelection, addressesByContractName, } + let chunkQuery = makeQuery(~isChunk=true) - let runPartialResponse = (~itemsCount, ~eventDensity=Some(1.)) => { + let runPartialResponse = (~itemsCount, ~eventDensity=Some(1.), ~isChunk=true) => { + let query = makeQuery(~isChunk) let fetchState = makeFetchState(~eventDensity) - fetchState->FetchState.startFetchingQueries(~queries=[chunkQuery]) + fetchState->FetchState.startFetchingQueries(~queries=[query]) let updated = fetchState->FetchState.handleQueryResult( - ~query=chunkQuery, + ~query, ~latestFetchedBlock={blockNumber: 90, blockTimestamp: 90 * 15}, ~newItems=Array.fromInitializer(~length=itemsCount, i => mockEvent(~blockNumber=10, ~logIndex=i) @@ -4796,15 +4799,26 @@ describe("Response density and source range capacity update independently", () = } it("updates density on every response but preserves capacity on a cap hit", t => { + // A non-chunk bounded query still sends its itemsTarget cap, so a response + // hitting it (itemsCount == itemsTarget) preserves the 300 capacity history. t.expect({ - "capHit": runPartialResponse(~itemsCount=3), - "subCap": runPartialResponse(~itemsCount=2), + "capHit": runPartialResponse(~itemsCount=3, ~isChunk=false), + "subCap": runPartialResponse(~itemsCount=2, ~isChunk=false), }).toEqual({ "capHit": (Some(300), Some((1. +. 3. /. 90.) /. 2.)), "subCap": (Some(90), Some((1. +. 2. /. 90.) /. 2.)), }) }) + it("trusts a bounded chunk's partial response as capacity — it sends no cap", t => { + // Bounded chunks carry no server cap, so a partial response is genuine + // capacity evidence even when itemsCount reaches what a cap would have been. + t.expect(runPartialResponse(~itemsCount=3, ~isChunk=true)).toEqual(( + Some(90), + Some((1. +. 3. /. 90.) /. 2.), + )) + }) + it("trusts cap-hit density before source capacity is known", t => { let fetchState = makeFetchState(~eventDensity=None, ~sourceRangeCapacity=0) fetchState->FetchState.startFetchingQueries(~queries=[chunkQuery]) From 10cead78accee67cede3601b340a4d687137deb5 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 24 Jul 2026 18:14:39 +0000 Subject: [PATCH 2/2] Make query itemsTarget an option; size everything on itemsEst Follow-up to dropping the server cap for bounded chunk queries. Every chunk/gap-fill query is bounded (its toBlock is sized to the source's range capacity), so it now carries itemsTarget None and sends no server cap; only open-ended probes, whose range isn't otherwise bounded, keep a cap (itemsTarget Some(itemsEst)). query.itemsTarget becomes option and SourceManager forwards it straight through. itemsEst is now the sole query-sizing and budget-reservation unit. This retires the chunkItemsMultiplier headroom and the itemsTargetFloor: both only ever sized a bounded query's server cap, never the itemsEst-based reservation, so with bounded queries uncapped they no longer do anything. Drops the params from FetchState.getNextQuery, ChainState.getNextQuery, and CrossChainState's 1.5x/3x realtime headroom. Tests updated to the option representation and to assert itemsEst where they previously asserted the (now-absent) cap. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01GqyCTGE239bpYGZ9UvfJDn --- packages/envio/src/ChainState.res | 19 +- packages/envio/src/ChainState.resi | 8 +- packages/envio/src/CrossChainState.res | 22 +-- packages/envio/src/FetchState.res | 110 ++++------- packages/envio/src/sources/SourceManager.res | 6 +- .../test_codegen/test/IndexerState_test.res | 8 +- .../test/lib_tests/CrossChainState_test.res | 29 ++- .../lib_tests/FetchState_onBlock_test.res | 12 +- .../test/lib_tests/FetchState_test.res | 174 +++++++++--------- .../test/lib_tests/SourceManager_test.res | 14 +- 10 files changed, 159 insertions(+), 243 deletions(-) diff --git a/packages/envio/src/ChainState.res b/packages/envio/src/ChainState.res index 5652e7133..fddbefa48 100644 --- a/packages/envio/src/ChainState.res +++ b/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,10 +561,6 @@ 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) @@ -578,12 +568,7 @@ let getNextQuery = ( // 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 diff --git a/packages/envio/src/ChainState.resi b/packages/envio/src/ChainState.resi index b603943ee..c79bd67ef 100644 --- a/packages/envio/src/ChainState.resi +++ b/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, diff --git a/packages/envio/src/CrossChainState.res b/packages/envio/src/CrossChainState.res index 0b6cbdb67..a2ba58d6a 100644 --- a/packages/envio/src/CrossChainState.res +++ b/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, }, ) ) diff --git a/packages/envio/src/FetchState.res b/packages/envio/src/FetchState.res index e74ca1eb5..9c1ca0dd7 100644 --- a/packages/envio/src/FetchState.res +++ b/packages/envio/src/FetchState.res @@ -25,11 +25,11 @@ type pendingQuery = { fromBlock: int, toBlock: option, isChunk: bool, - // Items this in-flight query is targeting (server maxNumLogs-style cap). - itemsTarget: int, - // Estimated items this in-flight query will actually return (no headroom), - // carried from the query so the shared buffer budget can account for what's - // already being fetched without the cap's safety margin inflating it. + // Server maxNumLogs-style cap for this in-flight query. None for bounded + // chunk/gap-fill queries, which send no cap (their toBlock is the bound). + itemsTarget: option, + // Estimated items this in-flight query will return, carried from the query so + // the shared buffer budget can account for what's already being fetched. itemsEst: int, // Stores latestFetchedBlock when query completes. Only needed to persist // timestamp while earlier queries are still pending before updating @@ -81,14 +81,15 @@ type query = { fromBlock: int, toBlock: option, isChunk: bool, - // Items this query targets: the server-side maxNumLogs-style cap, sized - // with headroom (chunkItemsMultiplier) so a denser-than-expected range - // doesn't truncate the response. - itemsTarget: int, - // Expected items without headroom: density × the query's block range for a - // known-density partition, the query's budget share otherwise. This is the - // unit the chain's per-tick budget is reserved/consumed in — reserving the - // headroomed cap instead would throttle the pipeline by the safety margin. + // Server-side maxNumLogs-style cap. Some only for open-ended probes, whose + // range isn't otherwise bounded; None for bounded chunk/gap-fill queries, + // whose toBlock already bounds the response so a cap would only self-truncate + // (worse under client-side address filtering, which counts filtered-out items + // toward the cap). + itemsTarget: option, + // Density × the query's block range for a known-density partition, the + // query's budget share otherwise. This is the unit the chain's per-tick + // budget is reserved/consumed in and that sizes every query. itemsEst: int, selection: selection, addressesByContractName: dict>, @@ -112,13 +113,13 @@ let deriveContractNameByAddress: dict> => dict< result }) -// itemsTarget for a query over [fromBlock, toBlock] at the given event density +// itemsEst for a query over [fromBlock, toBlock] at the given event density // (items/block). toBlock None is the open-ended tail, capped at // chainTargetBlock — the soft per-tick horizon the owning chain wants to reach // (see getNextQuery). let densityItemsTarget = (~density, ~fromBlock, ~toBlock, ~chainTargetBlock) => { - // Floor at 1: the reservation must equal the server-side cap SourceManager - // sends, and a 0 cap would ask the backend for nothing. + // Floor at 1: this is the budget reservation, and for a probe also its server + // cap — a 0 cap would ask the backend for nothing. Pervasives.max( 1, ((toBlock->Option.getOr(chainTargetBlock) - fromBlock + 1)->Int.toFloat *. density) @@ -548,10 +549,12 @@ module OptimizedPartitions = { if latestFetchedBlock.blockNumber < queryToBlock { // Partial response is direct capacity evidence — unless it was // truncated by our own itemsTarget cap: that reflects the - // reservation we asked for, not what the server could return. - // Bounded chunks send no cap (see SourceManager), so their partial - // responses are always genuine source-capacity evidence. - query.isChunk || itemsCount < query.itemsTarget + // reservation we asked for, not what the server could return. A + // capless bounded query can only have been truncated by the source. + switch query.itemsTarget { + | None => true + | Some(itemsTarget) => itemsCount < itemsTarget + } } else { // A full response updates only when the query's intended range // covers at least the partition's current chunk range — meaning it @@ -1813,19 +1816,12 @@ let maxChainConcurrency = Env.maxChainConcurrency // the first measurement. let chunkRangeGrowthFactor = 1.8 -// Push one density-priced query and return its itemsEst. itemsEst is the honest -// density estimate the chain's budget is reserved in; itemsTarget is the -// server-side cap with chunkItemsMultiplier headroom so a denser-than-expected -// range doesn't truncate the response. -// -// A bounded query (toBlock set) additionally floors its cap at -// itemsTargetFloor: its range is already the hard bound on the response, so a -// low density estimate shrinking the cap only buys self-truncated responses — -// each one a wasted roundtrip that opens a gap and pollutes nothing but our -// own pipeline. The floor is the indexer target split across the chain's -// concurrency slots, so even every in-flight bounded query hitting its floored -// cap at once overshoots the pool by at most ~one buffer target. Open-ended -// queries keep the pure density cap — there it's the only bound at all. +// Push one density-priced query and return its itemsEst, the density estimate +// that sizes the query and the chain's budget reservation. These queries are +// chunks and gap-fills: their toBlock is a tight bound sized to the source's +// range capacity, so they send no server cap (itemsTarget None) — a cap would +// only self-truncate the bounded range, worse under client-side address +// filtering. A rare unbounded call caps at the estimate as its only bound. let pushDensityPricedQuery = ( queries: array, ~partitionId, @@ -1833,19 +1829,11 @@ let pushDensityPricedQuery = ( ~toBlock, ~isChunk, ~density, - ~chunkItemsMultiplier, ~chainTargetBlock, - ~itemsTargetFloor, ~selection, ~addressesByContractName, ) => { let itemsEst = densityItemsTarget(~density, ~fromBlock, ~toBlock, ~chainTargetBlock) - let itemsTarget = densityItemsTarget( - ~density=density *. chunkItemsMultiplier, - ~fromBlock, - ~toBlock, - ~chainTargetBlock, - ) queries ->Array.push({ partitionId, @@ -1854,8 +1842,8 @@ let pushDensityPricedQuery = ( selection, isChunk, itemsTarget: switch toBlock { - | Some(_) => Pervasives.max(itemsTarget, itemsTargetFloor) - | None => itemsTarget + | Some(_) => None + | None => Some(itemsEst) }, itemsEst, addressesByContractName, @@ -1886,8 +1874,6 @@ let pushGapFillQueries = ( ~maxChunks: int, ~partition: partition, ~partitionBudget: float, - ~chunkItemsMultiplier: float, - ~itemsTargetFloor: int, ~selection: selection, ~addressesByContractName: dict>, ) => { @@ -1911,9 +1897,7 @@ let pushGapFillQueries = ( ~toBlock=rangeEndBlock, ~isChunk, ~density, - ~chunkItemsMultiplier, ~chainTargetBlock, - ~itemsTargetFloor, ~selection, ~addressesByContractName, ) @@ -1935,9 +1919,7 @@ let pushGapFillQueries = ( ~toBlock=Some(chunkToBlock), ~isChunk=true, ~density, - ~chunkItemsMultiplier, ~chainTargetBlock, - ~itemsTargetFloor, ~selection, ~addressesByContractName, ) @@ -1985,8 +1967,6 @@ let walkPartitionPending = ( ~candidates: array, ~headBlockNumber: int, ~chainTargetBlock: int, - ~chunkItemsMultiplier: float, - ~itemsTargetFloor: int, ~partitionBudget: float, ~queryEndBlock: option, ): option => { @@ -2013,8 +1993,6 @@ let walkPartitionPending = ( ~maxChunks=maxInFlightChunksPerPartition - inFlightCount - chunksUsedThisCall.contents, ~partition=p, ~partitionBudget, - ~chunkItemsMultiplier, - ~itemsTargetFloor, ~selection=p.selection, ~addressesByContractName=p.addressesByContractName, ) @@ -2058,14 +2036,12 @@ let pushForwardCandidates = ( // bound, see getNextQuery. ~inRangeStates: array, // The full in-range partition count, pre-truncation. Probe sizing divides by - // this so each probe's itemsEst/itemsTarget stays the honest per-partition - // share for budget control — sizing by the (fewer) admittable queries would - // let every accepted probe over-fetch its share. + // this so each probe's itemsEst stays the honest per-partition share for + // budget control — sizing by the (fewer) admittable queries would let every + // accepted probe over-fetch its share. ~inRangeCount: int, ~chainTargetBlock: int, ~freshBudget: float, - ~chunkItemsMultiplier: float, - ~itemsTargetFloor: int, ) => { // Even share of the fresh budget across the partitions actually fetching // this tick (not every partition — so budget isn't stranded on ones below @@ -2119,9 +2095,7 @@ let pushForwardCandidates = ( ~toBlock=Some(chunkToBlock), ~isChunk=true, ~density, - ~chunkItemsMultiplier, ~chainTargetBlock, - ~itemsTargetFloor, ~selection=p.selection, ~addressesByContractName=p.addressesByContractName, ) @@ -2135,7 +2109,7 @@ let pushForwardCandidates = ( // across the partitions fetching this tick. With no range to the target // fall back to an even share of the fresh budget, so cold chains and // caught-up partitions still probe. - let itemsTarget = if rangeToTarget > 0 { + let itemsEst = if rangeToTarget > 0 { Pervasives.max( 1, Math.round( @@ -2154,8 +2128,10 @@ let pushForwardCandidates = ( toBlock: fs.queryEndBlock, isChunk: false, selection: p.selection, - itemsTarget, - itemsEst: itemsTarget, + // An open-ended probe's range isn't bounded to source capacity, so it + // keeps a server cap at its estimate to protect the shared buffer. + itemsTarget: Some(itemsEst), + itemsEst, addressesByContractName: p.addressesByContractName, }) ->ignore @@ -2259,10 +2235,6 @@ let getNextQuery = ( {optimizedPartitions, blockLag, latestOnBlockBlockNumber, knownHeight, endBlock}: t, ~chainTargetBlock: int, ~chainTargetItems: float, - ~chunkItemsMultiplier: float=1., - // Floor for bounded queries' server cap (see pushDensityPricedQuery) — - // targetBufferSize / maxChainConcurrency from the cross-chain scheduler. - ~itemsTargetFloor: int=0, ) => { let headBlockNumber = knownHeight - blockLag if headBlockNumber <= 0 { @@ -2380,8 +2352,6 @@ let getNextQuery = ( ~candidates, ~headBlockNumber, ~chainTargetBlock, - ~chunkItemsMultiplier, - ~itemsTargetFloor, ~partitionBudget, ~queryEndBlock=computeQueryEndBlock(p), ) { @@ -2425,8 +2395,6 @@ let getNextQuery = ( ~inRangeCount, ~chainTargetBlock, ~freshBudget, - ~chunkItemsMultiplier, - ~itemsTargetFloor, ) acceptCandidates( diff --git a/packages/envio/src/sources/SourceManager.res b/packages/envio/src/sources/SourceManager.res index 6f0bacd67..f1123dfa0 100644 --- a/packages/envio/src/sources/SourceManager.res +++ b/packages/envio/src/sources/SourceManager.res @@ -768,11 +768,7 @@ let executeQuery = async ( ~partitionId=query.partitionId, ~knownHeight, ~selection=query.selection, - // Bounded chunk queries send no server cap: their toBlock is already the - // hard bound, and with client-side address filtering the server counts - // filtered-out items toward the cap, truncating the range short and - // opening gaps. Everything else keeps its cap. - ~itemsTarget=query.isChunk && query.toBlock->Option.isSome ? None : Some(query.itemsTarget), + ~itemsTarget=query.itemsTarget, ~retry, ~logger, ) diff --git a/scenarios/test_codegen/test/IndexerState_test.res b/scenarios/test_codegen/test/IndexerState_test.res index 631c04359..7330ffd85 100644 --- a/scenarios/test_codegen/test/IndexerState_test.res +++ b/scenarios/test_codegen/test/IndexerState_test.res @@ -7,7 +7,7 @@ let defaultQuery: FetchState.query = { fromBlock: 0, toBlock: None, isChunk: false, - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, selection: {FetchState.dependsOnAddresses: false, onEventRegistrations: []}, addressesByContractName: Dict.make(), @@ -90,7 +90,7 @@ let populateChainQueuesWithRandomEvents = (~runTime=1000, ~maxBlockTime=15, ()) let query: FetchState.query = { partitionId: "0", - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, fromBlock: 0, toBlock: None, @@ -278,7 +278,7 @@ describe("IndexerState", () => { blockNumber => { let query: FetchState.query = { partitionId: "0", - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, fromBlock: 0, toBlock: None, @@ -361,7 +361,7 @@ describe("IndexerState", () => { let cs = state->IndexerState.getChainState(~chain) let concurrentQuery: FetchState.query = { partitionId: "0", - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, fromBlock: 0, toBlock: None, diff --git a/scenarios/test_codegen/test/lib_tests/CrossChainState_test.res b/scenarios/test_codegen/test/lib_tests/CrossChainState_test.res index 186f34c79..81f1c992d 100644 --- a/scenarios/test_codegen/test/lib_tests/CrossChainState_test.res +++ b/scenarios/test_codegen/test/lib_tests/CrossChainState_test.res @@ -422,11 +422,10 @@ describe("CrossChainState fetch control", () => { // Chain 1 (furthest behind, so priorityOrder visits it first): a single // known-density partition with a short remaining range (endBlock=20 at - // density 10 items/block -> 200 items across 2 chunks, reserved at the - // chunk headroom multiplier: 1.5x backfill, 3x realtime). Its real + // density 10 items/block -> 200 items across 2 chunks). Its real // consumption is capped by that range, far below whatever share of the // 3000-item pool the waterfall would otherwise hand it. Returns each - // chain's dispatched itemsTarget total and pendingBudget. + // chain's dispatched itemsEst total and pendingBudget. let runShortRangeWaterfall = async (~isRealtime) => { let normalSelection = {FetchState.dependsOnAddresses: false, onEventRegistrations: []} let address1 = "0x1111111111111111111111111111111111111111"->Address.unsafeFromString @@ -510,7 +509,7 @@ describe("CrossChainState fetch control", () => { chain->ChainMap.Chain.toChainId, switch action { | Ready(queries) => - queries->Array.reduce(0., (acc, q: FetchState.query) => acc +. q.itemsTarget->Int.toFloat) + queries->Array.reduce(0., (acc, q: FetchState.query) => acc +. q.itemsEst->Int.toFloat) | _ => 0. }, ) @@ -530,16 +529,16 @@ describe("CrossChainState fetch control", () => { async t => { t.expect( await runShortRangeWaterfall(~isRealtime=false), - ~message="Chain 1's real range caps it at 200 items (itemsTarget carries the 1.5x headroom = 300, only the 200 estimate is reserved); chain 2's frontier is already past the alignment line anchored at chain 1's frontier, so it waits instead of draining the pool", - ).toEqual((Some(300.), Some(0.), 200., 0.)) + ~message="Chain 1's real range caps it at its honest 200-item estimate, which is also what pendingBudget reserves; chain 2's frontier is already past the alignment line anchored at chain 1's frontier, so it waits instead of draining the pool", + ).toEqual((Some(200.), Some(0.), 200., 0.)) }, ) - Async.it("checkAndFetch drops the alignment clamp in realtime and sizes chunk caps with 3x headroom", async t => { + Async.it("checkAndFetch drops the alignment clamp in realtime", async t => { t.expect( await runShortRangeWaterfall(~isRealtime=true), - ~message="Chain 1's itemsTarget carries the 3x realtime headroom = 600, but pendingBudget still reserves the honest 200-item estimate; chain 2 is unclamped at realtime and gets the rest", - ).toEqual((Some(600.), Some(2800.), 200., 2800.)) + ~message="Chain 1 is sized to its honest 200-item range estimate, which pendingBudget also reserves; chain 2 is unclamped at realtime and gets the rest", + ).toEqual((Some(200.), Some(2800.), 200., 2800.)) }) Async.it( @@ -563,7 +562,7 @@ describe("CrossChainState fetch control", () => { | Ready(queries) => "ready:" ++ queries - ->Array.reduce(0., (acc, q: FetchState.query) => acc +. q.itemsTarget->Int.toFloat) + ->Array.reduce(0., (acc, q: FetchState.query) => acc +. q.itemsEst->Int.toFloat) ->Float.toString }, ) @@ -634,13 +633,11 @@ describe("CrossChainState fetch control", () => { ) let itemsTarget = cs => switch cs->ChainState.getNextQuery(~chainTargetItems=3000.) { - | Ready([q]) => q.itemsTarget + | Ready([q]) => q.itemsEst | _ => JsError.throwWithMessage("expected a single ready query") } // Range cost to the 20-block endBlock ceiling at density 10 = 200 items. - // No extra headroom on the budget cap: truncation safety lives in the - // itemsTarget server cap via chunkItemsMultiplier, not in the reservation. t.expect( (makeChain(~caughtUpOnce=false)->itemsTarget, makeChain(~caughtUpOnce=true)->itemsTarget), ~message="Both are capped at the plain range cost", @@ -739,7 +736,7 @@ describe("ChainState cold start", () => { chain->ChainMap.Chain.toChainId, switch action { | Ready(queries) => - queries->Array.reduce(0., (acc, q: FetchState.query) => acc +. q.itemsTarget->Int.toFloat) + queries->Array.reduce(0., (acc, q: FetchState.query) => acc +. q.itemsEst->Int.toFloat) | _ => 0. }, ) @@ -782,7 +779,7 @@ describe("ChainState cold start", () => { | Ready(queries) => "ready:" ++ queries - ->Array.reduce(0., (acc, q: FetchState.query) => acc +. q.itemsTarget->Int.toFloat) + ->Array.reduce(0., (acc, q: FetchState.query) => acc +. q.itemsEst->Int.toFloat) ->Float.toString }, ) @@ -846,7 +843,7 @@ describe("ChainState cold start", () => { | Ready(queries) => dispatched := queries->Array.reduce(0., (acc, q: FetchState.query) => - acc +. q.itemsTarget->Int.toFloat + acc +. q.itemsEst->Int.toFloat ) | _ => () } diff --git a/scenarios/test_codegen/test/lib_tests/FetchState_onBlock_test.res b/scenarios/test_codegen/test/lib_tests/FetchState_onBlock_test.res index 415ab4d6d..98b063457 100644 --- a/scenarios/test_codegen/test/lib_tests/FetchState_onBlock_test.res +++ b/scenarios/test_codegen/test/lib_tests/FetchState_onBlock_test.res @@ -9,7 +9,7 @@ let defaultQuery: FetchState.query = { fromBlock: 0, toBlock: None, isChunk: false, - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, selection: {FetchState.dependsOnAddresses: false, onEventRegistrations: []}, addressesByContractName: Dict.make(), @@ -93,7 +93,7 @@ describe("FetchState onBlock functionality", () => { // This should trigger the onBlock logic and add block items to the queue let query: FetchState.query = { partitionId: "0", - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, toBlock: None, isChunk: false, @@ -141,7 +141,7 @@ describe("FetchState onBlock functionality", () => { // Process a batch that goes from block 0 to 10 let query: FetchState.query = { partitionId: "0", - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, toBlock: None, isChunk: false, @@ -190,7 +190,7 @@ describe("FetchState onBlock functionality", () => { // Process a batch that goes from block 0 to 10 let query: FetchState.query = { partitionId: "0", - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, toBlock: None, isChunk: false, @@ -243,7 +243,7 @@ describe("FetchState onBlock functionality", () => { // Process a batch let query: FetchState.query = { partitionId: "0", - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, toBlock: None, isChunk: false, @@ -299,7 +299,7 @@ describe("FetchState onBlock functionality", () => { // Process a batch let query: FetchState.query = { partitionId: "0", - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, toBlock: None, isChunk: false, diff --git a/scenarios/test_codegen/test/lib_tests/FetchState_test.res b/scenarios/test_codegen/test/lib_tests/FetchState_test.res index d1426012e..27cb1dd16 100644 --- a/scenarios/test_codegen/test/lib_tests/FetchState_test.res +++ b/scenarios/test_codegen/test/lib_tests/FetchState_test.res @@ -11,7 +11,7 @@ let defaultQuery: FetchState.query = { fromBlock: 0, toBlock: None, isChunk: false, - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, selection: {FetchState.dependsOnAddresses: false, onEventRegistrations: []}, addressesByContractName: Dict.make(), @@ -1893,7 +1893,7 @@ describe("FetchState.getNextQuery & integration", () => { Ready([ { partitionId: "0", - itemsTarget: 10000, + itemsTarget: Some(10000), itemsEst: 10000, fromBlock: 0, toBlock: None, @@ -1919,7 +1919,7 @@ describe("FetchState.getNextQuery & integration", () => { fromBlock: 0, toBlock: None, isChunk: false, - itemsTarget: 10000, + itemsTarget: Some(10000), itemsEst: 10000, fetchedBlock: None, }, @@ -1977,7 +1977,7 @@ describe("FetchState.getNextQuery & integration", () => { Ready([ { partitionId: "0", - itemsTarget: 10000, + itemsTarget: Some(10000), itemsEst: 10000, fromBlock: 11, toBlock: None, @@ -2012,7 +2012,7 @@ describe("FetchState.getNextQuery & integration", () => { Ready([ { partitionId: "0", - itemsTarget: 10000, + itemsTarget: Some(10000), itemsEst: 10000, toBlock: Some(8), selection: fetchState.normalSelection, @@ -2031,7 +2031,7 @@ describe("FetchState.getNextQuery & integration", () => { Ready([ { partitionId: "0", - itemsTarget: 10000, + itemsTarget: Some(10000), itemsEst: 10000, toBlock: Some(8), selection: fetchState.normalSelection, @@ -2129,7 +2129,7 @@ describe("FetchState.getNextQuery & integration", () => { Ready([ { partitionId: "1", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, toBlock: Some(10), isChunk: false, @@ -2141,7 +2141,7 @@ describe("FetchState.getNextQuery & integration", () => { partitionId: "2", // Sits one block ahead of partition "1", so 9/10 of the range to the // target -> 4500 vs 5000. - itemsTarget: 4500, + itemsTarget: Some(4500), itemsEst: 4500, fromBlock: 2, toBlock: None, @@ -2188,20 +2188,20 @@ describe("FetchState.getNextQuery & integration", () => { makeIntermidiateDcMerge(), ) - let makePartition2Query = (~itemsTarget): FetchState.query => { + let makePartition2Query = (~itemsEst): FetchState.query => { partitionId: "2", - itemsTarget, - itemsEst: itemsTarget, + itemsTarget: Some(itemsEst), + itemsEst, fromBlock: 3, toBlock: None, selection: fetchState.normalSelection, addressesByContractName: Dict.fromArray([("Gravatar", [mockAddress3])]), isChunk: false, } - let makePartition0Query = (~itemsTarget): FetchState.query => { + let makePartition0Query = (~itemsEst): FetchState.query => { partitionId: "0", - itemsTarget, - itemsEst: itemsTarget, + itemsTarget: Some(itemsEst), + itemsEst, toBlock: None, selection: fetchState.normalSelection, addressesByContractName: Dict.fromArray([ @@ -2219,11 +2219,11 @@ describe("FetchState.getNextQuery & integration", () => { ).toEqual( // Partition "0" sits at block 11 (the head), covering only the last block // of the range to the target -> a small probe next to "2"'s 5000. - Ready([makePartition2Query(~itemsTarget=5000), makePartition0Query(~itemsTarget=556)]), + Ready([makePartition2Query(~itemsEst=5000), makePartition0Query(~itemsEst=556)]), ) // Partition "0" is above the target block, so it's the only eligible // unknown-density partition here and gets the whole budget. - let partition2QuerySolo = makePartition2Query(~itemsTarget=10000) + let partition2QuerySolo = makePartition2Query(~itemsEst=10000) t.expect( updatedFetchState->getNextQuery(~knownHeight=10), ~message=`Even if a single partition reached block height, @@ -2239,7 +2239,7 @@ describe("FetchState.getNextQuery & integration", () => { t.expect( updatedFetchState->getNextQuery(~knownHeight=11, ~chainTargetItems=20_000.), ~message=`Should skip fetching queries`, - ).toEqual(Ready([makePartition0Query(~itemsTarget=10000)])) + ).toEqual(Ready([makePartition0Query(~itemsEst=10000)])) }) it("Emulate partition merging cases", t => { @@ -2258,7 +2258,7 @@ describe("FetchState.getNextQuery & integration", () => { Ready([ { partitionId: "2", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, toBlock: None, selection: originalFetchState.normalSelection, @@ -2270,7 +2270,7 @@ describe("FetchState.getNextQuery & integration", () => { FetchState.partitionId: "0", // At block 11 (the head), it covers only the last block of the range // to the target, so a small probe next to partition "2"'s 5000. - itemsTarget: 556, + itemsTarget: Some(556), itemsEst: 556, toBlock: None, selection: originalFetchState.normalSelection, @@ -2293,7 +2293,7 @@ describe("FetchState.getNextQuery & integration", () => { Ready([ { partitionId: "2", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, toBlock: Some(10), selection: fetchState.normalSelection, @@ -2303,7 +2303,7 @@ describe("FetchState.getNextQuery & integration", () => { }, { FetchState.partitionId: "0", - itemsTarget: 556, + itemsTarget: Some(556), itemsEst: 556, toBlock: None, selection: originalFetchState.normalSelection, @@ -2382,7 +2382,7 @@ describe("FetchState.getNextQuery & integration", () => { fromBlock: 11, toBlock: None, isChunk: false, - itemsTarget: 2500, + itemsTarget: Some(2500), itemsEst: 2500, fetchedBlock: None, }, @@ -2452,7 +2452,7 @@ describe("FetchState.getNextQuery & integration", () => { Ready([ { partitionId: "0", - itemsTarget: 3333, + itemsTarget: Some(3333), itemsEst: 3333, fromBlock: 0, toBlock: None, @@ -2465,7 +2465,7 @@ describe("FetchState.getNextQuery & integration", () => { }, { partitionId: "1", - itemsTarget: 3333, + itemsTarget: Some(3333), itemsEst: 3333, fromBlock: 0, toBlock: None, @@ -2476,7 +2476,7 @@ describe("FetchState.getNextQuery & integration", () => { { partitionId: "2", // Starts at block 2, so 9 of the 11-block range to the target -> 2727. - itemsTarget: 2727, + itemsTarget: Some(2727), itemsEst: 2727, fromBlock: 2, toBlock: None, @@ -2652,7 +2652,7 @@ describe("FetchState.getNextQuery & integration", () => { ~queries=[ { partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, toBlock: None, selection: { @@ -2769,7 +2769,7 @@ describe("FetchState unit tests for specific cases", () => { let query: FetchState.query = { partitionId: "1", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, fromBlock: 1, toBlock: None, @@ -2815,7 +2815,7 @@ describe("FetchState unit tests for specific cases", () => { let query: FetchState.query = { partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, fromBlock: 0, toBlock: None, @@ -2868,7 +2868,7 @@ describe("FetchState unit tests for specific cases", () => { let query0: FetchState.query = { partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, fromBlock: 0, toBlock: None, @@ -2881,7 +2881,7 @@ describe("FetchState unit tests for specific cases", () => { } let query1: FetchState.query = { partitionId: "1", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, fromBlock: 0, toBlock: None, @@ -2915,7 +2915,7 @@ describe("FetchState unit tests for specific cases", () => { Ready([ { partitionId: "0", - itemsTarget: 10000, + itemsTarget: Some(10000), itemsEst: 10000, fromBlock: 2, toBlock: None, @@ -2946,7 +2946,7 @@ describe("FetchState unit tests for specific cases", () => { let query: FetchState.query = { partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, fromBlock: 0, toBlock: None, @@ -3013,7 +3013,7 @@ describe("FetchState unit tests for specific cases", () => { let query: FetchState.query = { partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, fromBlock: 0, toBlock: None, @@ -3130,7 +3130,7 @@ describe("FetchState unit tests for specific cases", () => { } let query: FetchState.query = { partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, fromBlock: 0, toBlock: Some(0), @@ -3155,7 +3155,7 @@ describe("FetchState unit tests for specific cases", () => { let fetchToHead = (fetchState: FetchState.t, ~latestFetchedBlockNumber) => { let query: FetchState.query = { partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, fromBlock: 0, toBlock: None, @@ -3211,7 +3211,7 @@ describe("FetchState unit tests for specific cases", () => { let query: FetchState.query = { partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, selection: fetchState.normalSelection, addressesByContractName: Dict.fromArray([("Gravatar", [mockAddress1])]), @@ -3307,7 +3307,7 @@ describe("FetchState unit tests for specific cases", () => { // Partition "1" is back in range now that its query resolved, so // the even split is now 3-way instead of 2-way. ...partition2Query, - itemsTarget: 3333, + itemsTarget: Some(3333), itemsEst: 3333, }, { @@ -3317,7 +3317,7 @@ describe("FetchState unit tests for specific cases", () => { // less of the range to the target and gets a smaller probe. ...queryA, partitionId: "1", - itemsTarget: 1663, + itemsTarget: Some(1663), itemsEst: 1663, toBlock: Some(500), fromBlock: 401, @@ -3326,7 +3326,7 @@ describe("FetchState unit tests for specific cases", () => { // Partition "0" starts even further ahead (block 501), so it covers // the least range and gets the smallest probe. ...queries->Array.getUnsafe(1), - itemsTarget: 831, + itemsTarget: Some(831), itemsEst: 831, }, ]), @@ -3339,7 +3339,7 @@ describe("FetchState.sortForBatch", () => { let mkQuery = (fetchState: FetchState.t) => { { FetchState.partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, toBlock: None, isChunk: false, @@ -3711,7 +3711,7 @@ describe("FetchState progress tracking", () => { let (fs0, _) = makeInitial(~knownHeight=1000) let query = { FetchState.partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, toBlock: None, isChunk: false, @@ -3786,7 +3786,7 @@ describe("FetchState proposes queries against the natural ceiling", () => { let query0 = { FetchState.partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, toBlock: None, isChunk: false, @@ -3839,7 +3839,7 @@ describe("FetchState proposes queries against the natural ceiling", () => { // Test case 3: Small queue -> Should also use the open-ended head target let query3 = { FetchState.partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, toBlock: None, isChunk: false, @@ -4167,7 +4167,7 @@ describe("FetchState.getNextQuery water-fill round is order-independent", () => let getItemsTargetByPartition = nextQuery => switch nextQuery { | FetchState.Ready(queries) => - queries->Array.map((q: FetchState.query) => (q.partitionId, q.itemsTarget)) + queries->Array.map((q: FetchState.query) => (q.partitionId, q.itemsEst)) | _ => [] } @@ -4251,8 +4251,8 @@ describe("FetchState.getNextQuery greedy budget pass fills partitions toward the | Ready(queries) => queries->Array.forEach((q: FetchState.query) => switch byPartition->Dict.get(q.partitionId) { - | Some(arr) => arr->Array.push((q.fromBlock, q.itemsTarget))->ignore - | None => byPartition->Dict.set(q.partitionId, [(q.fromBlock, q.itemsTarget)]) + | Some(arr) => arr->Array.push((q.fromBlock, q.itemsEst))->ignore + | None => byPartition->Dict.set(q.partitionId, [(q.fromBlock, q.itemsEst)]) } ) | _ => () @@ -4292,7 +4292,7 @@ describe("FetchState.getNextQuery with uneven in-flight reservations", () => { fromBlock: 1, toBlock: Some(100), isChunk: true, - itemsTarget, + itemsTarget: None, itemsEst: itemsTarget, fetchedBlock: None, }, @@ -4343,8 +4343,8 @@ describe("FetchState.getNextQuery with uneven in-flight reservations", () => { | Ready(queries) => queries->Array.forEach((q: FetchState.query) => switch byPartition->Dict.get(q.partitionId) { - | Some(arr) => arr->Array.push((q.fromBlock, q.itemsTarget))->ignore - | None => byPartition->Dict.set(q.partitionId, [(q.fromBlock, q.itemsTarget)]) + | Some(arr) => arr->Array.push((q.fromBlock, q.itemsEst))->ignore + | None => byPartition->Dict.set(q.partitionId, [(q.fromBlock, q.itemsEst)]) } ) | _ => () @@ -4384,7 +4384,7 @@ describe("FetchState.getNextQuery with uneven in-flight reservations", () => { fromBlock: 1, toBlock: None, isChunk: false, - itemsTarget: 250, + itemsTarget: Some(250), itemsEst: 250, selection: normalSelection, addressesByContractName: Dict.fromArray([("MockContract", [mockAddress0])]), @@ -4394,7 +4394,7 @@ describe("FetchState.getNextQuery with uneven in-flight reservations", () => { fromBlock: 101, toBlock: Some(118), isChunk: true, - itemsTarget: 180, + itemsTarget: None, itemsEst: 180, selection: normalSelection, addressesByContractName: Dict.fromArray([("MockContract", [mockAddress1])]), @@ -4404,7 +4404,7 @@ describe("FetchState.getNextQuery with uneven in-flight reservations", () => { fromBlock: 119, toBlock: Some(136), isChunk: true, - itemsTarget: 180, + itemsTarget: None, itemsEst: 180, selection: normalSelection, addressesByContractName: Dict.fromArray([("MockContract", [mockAddress1])]), @@ -4428,7 +4428,7 @@ describe("FetchState.getNextQuery with uneven in-flight reservations", () => { fromBlock: 1, toBlock: None, isChunk: false, - itemsTarget: 100, + itemsTarget: Some(100), itemsEst: 100, selection: normalSelection, addressesByContractName: Dict.fromArray([("MockContract", [address])]), @@ -4470,7 +4470,7 @@ describe("FetchState.getNextQuery with uneven in-flight reservations", () => { fromBlock: 1, toBlock: None, isChunk: false, - itemsTarget: 500, + itemsTarget: Some(500), itemsEst: 500, selection: normalSelection, addressesByContractName: Dict.fromArray([("MockContract", [mockAddress0])]), @@ -4480,7 +4480,7 @@ describe("FetchState.getNextQuery with uneven in-flight reservations", () => { fromBlock: 51, toBlock: None, isChunk: false, - itemsTarget: 250, + itemsTarget: Some(250), itemsEst: 250, selection: normalSelection, addressesByContractName: Dict.fromArray([("MockContract", [mockAddress1])]), @@ -4562,7 +4562,7 @@ describe("FetchState.getNextQuery target containment", () => { ~latestFetchedBlock=100, ~knownDensity=false, ~mutPendingQueries=[ - {fromBlock: 200, toBlock: Some(219), isChunk: true, itemsTarget: 100, itemsEst: 100, fetchedBlock: None}, + {fromBlock: 200, toBlock: Some(219), isChunk: true, itemsTarget: None, itemsEst: 100, fetchedBlock: None}, ], ), ) @@ -4598,7 +4598,7 @@ describe("FetchState.getNextQuery target containment", () => { fromBlock: 101, toBlock: Some(200), isChunk: true, - itemsTarget: 1500, + itemsTarget: None, itemsEst: 1500, fetchedBlock: Some({blockNumber: 200, blockTimestamp: 0}), }, @@ -4660,28 +4660,25 @@ describe("FetchState.getNextQuery chunk headroom and budget-driven emit", () => clientFilterAddressThreshold: None, } - let getChunks = (fetchState: FetchState.t, ~chainTargetItems, ~chunkItemsMultiplier=?) => + let getChunks = (fetchState: FetchState.t, ~chainTargetItems) => switch fetchState->FetchState.getNextQuery( ~chainTargetBlock=100000, ~chainTargetItems, - ~chunkItemsMultiplier?, ) { - | Ready(queries) => queries->Array.map((q: FetchState.query) => (q.fromBlock, q.itemsTarget)) + | Ready(queries) => queries->Array.map((q: FetchState.query) => (q.fromBlock, q.itemsEst)) | _ => [] } - it("sizes chunk itemsTarget with the chunk headroom multiplier", t => { - t.expect({ - "backfill1_5x": makeFetchState()->getChunks(~chainTargetItems=270., ~chunkItemsMultiplier=1.5), - "realtime3x": makeFetchState()->getChunks(~chainTargetItems=270., ~chunkItemsMultiplier=3.), - }).toEqual({ - // The 270-item budget is consumed in honest 180-item estimates: the - // first chunk leaves 90, which re-pours and forces a second full chunk. - // ceil(1.5 * 10 * 18) = 270 per chunk. - "backfill1_5x": [(1, 270), (19, 270)], - // ceil(3 * 10 * 18) = 540 per chunk. - "realtime3x": [(1, 540), (19, 540)], - }) + it("chunk queries carry no server cap (itemsTarget: None)", t => { + let chunkCaps = switch makeFetchState()->FetchState.getNextQuery( + ~chainTargetBlock=100000, + ~chainTargetItems=270., + ) { + | Ready(queries) => queries->Array.map((q: FetchState.query) => (q.fromBlock, q.itemsTarget)) + | _ => [] + } + // A chunk is bounded by its toBlock, so it sends no server-side item cap. + t.expect(chunkCaps).toEqual([(1, None), (19, None)]) }) it("emits chunks while the budget lasts, min one chunk per water-fill round", t => { @@ -4697,28 +4694,25 @@ describe("FetchState.getNextQuery chunk headroom and budget-driven emit", () => }) }) - it("floors bounded chunk caps at itemsTargetFloor, leaving open-ended probes untouched", t => { - let getQueries = (fetchState: FetchState.t, ~chainTargetItems) => + it("bounded chunks send no cap while the open-ended probe keeps its budget-share cap", t => { + let getCaps = (fetchState: FetchState.t, ~chainTargetItems) => switch fetchState->FetchState.getNextQuery( ~chainTargetBlock=100000, ~chainTargetItems, - ~itemsTargetFloor=1000, ) { | Ready(queries) => queries->Array.map((q: FetchState.query) => (q.fromBlock, q.itemsTarget)) | _ => [] } t.expect({ - "boundedChunks": makeFetchState()->getQueries(~chainTargetItems=400.), - "openProbe": makeFetchState(~eventDensity=None)->getQueries(~chainTargetItems=50.), + "boundedChunks": makeFetchState()->getCaps(~chainTargetItems=400.), + "openProbe": makeFetchState(~eventDensity=None)->getCaps(~chainTargetItems=50.), }).toEqual({ - // A chunk's range is already the hard bound on its response, so the - // floor lifts the 180-item density cap to 1000 — a low density estimate - // must not shrink caps into self-truncated responses. itemsEst stays at - // the honest 180, so budget acceptance still emits the same 3 chunks. - "boundedChunks": [(1, 1000), (19, 1000), (37, 1000)], + // A chunk's range is already the hard bound on its response, so it carries + // no server-side item cap. + "boundedChunks": [(1, None), (19, None), (37, None)], // The open-ended probe's cap is its only response bound, so it keeps its - // budget-share size instead of being floored. - "openProbe": [(1, 50)], + // budget-share size. + "openProbe": [(1, Some(50))], }) }) }) @@ -4775,7 +4769,7 @@ describe("Response density and source range capacity update independently", () = fromBlock: 1, toBlock: Some(540), isChunk, - itemsTarget: 3, + itemsTarget: isChunk ? None : Some(3), itemsEst: 3, selection: normalSelection, addressesByContractName, @@ -4896,7 +4890,7 @@ describe("FetchState.getNextQuery caps per-chain concurrency", () => { fromBlock: 1, toBlock: None, isChunk: false, - itemsTarget: 1, + itemsTarget: Some(1), itemsEst: 1, fetchedBlock: None, }, @@ -4969,7 +4963,7 @@ describe("FetchState.getNextQuery caps per-chain concurrency", () => { // control — even though the concurrency cap admits only 100 queries. t.expect({ "count": queries->Array.length, - "firstItemsTarget": (queries->Array.getUnsafe(0)).itemsTarget, + "firstItemsTarget": (queries->Array.getUnsafe(0)).itemsEst, }).toEqual({ "count": 100, "firstItemsTarget": 8333, @@ -4984,7 +4978,7 @@ describe("FetchState.getNextQuery caps per-chain concurrency", () => { FetchState.fromBlock: idx * 10 + 1, toBlock: Some((idx + 1) * 10), isChunk: true, - itemsTarget: 1, + itemsTarget: None, itemsEst: 1, fetchedBlock: idx === 0 ? None : Some({blockNumber: (idx + 1) * 10, blockTimestamp: 0}), }) @@ -5008,7 +5002,7 @@ describe("FetchState.getNextQuery caps per-chain concurrency", () => { toBlock: None, isChunk: false, selection: normalSelection, - itemsTarget: 999, + itemsTarget: Some(999), itemsEst: 999, addressesByContractName, }, @@ -5042,7 +5036,7 @@ describe("FetchState.getNextQuery caps per-chain concurrency", () => { toBlock: None, isChunk: false, selection: normalSelection, - itemsTarget: 10, + itemsTarget: Some(10), itemsEst: 10, addressesByContractName, } diff --git a/scenarios/test_codegen/test/lib_tests/SourceManager_test.res b/scenarios/test_codegen/test/lib_tests/SourceManager_test.res index 53a9c4b4a..bc607b6e4 100644 --- a/scenarios/test_codegen/test/lib_tests/SourceManager_test.res +++ b/scenarios/test_codegen/test/lib_tests/SourceManager_test.res @@ -7,7 +7,7 @@ let defaultQuery: FetchState.query = { fromBlock: 0, toBlock: None, isChunk: false, - itemsTarget: 0, + itemsTarget: Some(0), itemsEst: 0, selection: {FetchState.dependsOnAddresses: false, onEventRegistrations: []}, addressesByContractName: Dict.make(), @@ -180,7 +180,7 @@ describe("SourceManager source priority with Live sources", () => { let mockQuery = (): FetchState.query => { partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, fromBlock: 0, toBlock: None, @@ -489,7 +489,7 @@ describe("SourceManager fetchNext", () => { fromBlock: idx * 10 + 1, toBlock: Some(idx * 10 + 10), isChunk: true, - itemsTarget: 5000, + itemsTarget: None, itemsEst: 5000, fetchedBlock: None, } @@ -550,7 +550,7 @@ describe("SourceManager fetchNext", () => { ).toEqual([ { partitionId: "2", - itemsTarget: 16_667, + itemsTarget: Some(16_667), itemsEst: 16_667, fromBlock: 2, toBlock: None, @@ -562,7 +562,7 @@ describe("SourceManager fetchNext", () => { partitionId: "0", // Starts at block 5 vs partition "2"'s block 2, so it covers less of // the range to the target and gets a smaller probe. - itemsTarget: 11_111, + itemsTarget: Some(11_111), itemsEst: 11_111, fromBlock: 5, toBlock: None, @@ -573,7 +573,7 @@ describe("SourceManager fetchNext", () => { { partitionId: "1", // Starts furthest ahead (block 6), so it gets the smallest probe. - itemsTarget: 9_259, + itemsTarget: Some(9_259), itemsEst: 9_259, fromBlock: 6, toBlock: None, @@ -1440,7 +1440,7 @@ describe("SourceManager.executeQuery", () => { let mockQuery = (): FetchState.query => { partitionId: "0", - itemsTarget: 5000, + itemsTarget: Some(5000), itemsEst: 5000, fromBlock: 0, toBlock: None,