|
| 1 | +package ethrpc |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "math" |
| 6 | + "math/big" |
| 7 | + "strings" |
| 8 | + |
| 9 | + "github.com/0xsequence/ethkit/go-ethereum" |
| 10 | + "github.com/0xsequence/ethkit/go-ethereum/core/types" |
| 11 | +) |
| 12 | + |
| 13 | +// filterLogs executes a standard eth_getLogs JSON-RPC call for the given query. |
| 14 | +func (p *Provider) filterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error) { |
| 15 | + var logs []types.Log |
| 16 | + _, err := p.Do(ctx, FilterLogs(q).Strict(p.strictness).Into(&logs)) |
| 17 | + return logs, err |
| 18 | +} |
| 19 | + |
| 20 | +// filterLogsAutoSplit executes eth_getLogs with automatic range splitting when the node |
| 21 | +// returns a "too much data" style error. It uses an AIMD (Additive Increase, |
| 22 | +// Multiplicative Decrease) strategy to adapt the batch range over time. |
| 23 | +// |
| 24 | +// Calibration: on the first call the provider tries the full range (or the explicit max). |
| 25 | +// If the node rejects it, we shrink (multiplicative decrease /1.5) and retry. Within a |
| 26 | +// single call, after each successful sub-query we attempt additive increase (+10%). If |
| 27 | +// the increase triggers another error, we shrink back and mark the range as "calibrated" |
| 28 | +// — meaning the node's true limit has been found. Once calibrated, subsequent calls skip |
| 29 | +// the probing entirely and use the known-good range directly. |
| 30 | +func (p *Provider) filterLogsAutoSplit(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error) { |
| 31 | + from, to, err := p.resolveFilterBlockRange(ctx, q) |
| 32 | + if err != nil { |
| 33 | + return nil, err |
| 34 | + } |
| 35 | + |
| 36 | + // Single block: just do a direct call. Empty/invalid range: return early. |
| 37 | + if from == to { |
| 38 | + return p.filterLogs(ctx, q) |
| 39 | + } |
| 40 | + if from > to { |
| 41 | + return nil, nil |
| 42 | + } |
| 43 | + |
| 44 | + totalRange := to - from |
| 45 | + batchRange := p.effectiveFilterLogsBatchRange(totalRange) |
| 46 | + |
| 47 | + // Additive factor: 10% of the starting batch range, minimum 1 |
| 48 | + additiveFactor := uint64(math.Ceil(float64(batchRange) * 0.10)) |
| 49 | + if additiveFactor < 1 { |
| 50 | + additiveFactor = 1 |
| 51 | + } |
| 52 | + |
| 53 | + // Track whether we've succeeded after a shrink. We only mark calibrated |
| 54 | + // on a shrink→success→grow→fail cycle, which means we've bracketed the |
| 55 | + // true limit. A shrink→shrink sequence (consecutive failures) does not |
| 56 | + // count as calibration. |
| 57 | + succeededSinceShrink := false |
| 58 | + |
| 59 | + var allLogs []types.Log |
| 60 | + |
| 61 | + for cursor := from; cursor <= to; { |
| 62 | + if err := ctx.Err(); err != nil { |
| 63 | + return nil, err |
| 64 | + } |
| 65 | + |
| 66 | + end := cursor + batchRange |
| 67 | + if end > to { |
| 68 | + end = to |
| 69 | + } |
| 70 | + |
| 71 | + subQ := q // copy value (FilterQuery is a struct) |
| 72 | + subQ.FromBlock = new(big.Int).SetUint64(cursor) |
| 73 | + subQ.ToBlock = new(big.Int).SetUint64(end) |
| 74 | + |
| 75 | + logs, err := p.filterLogs(ctx, subQ) |
| 76 | + if err != nil { |
| 77 | + if !isFilterLogsTooMuchDataError(err) { |
| 78 | + return nil, err |
| 79 | + } |
| 80 | + |
| 81 | + // If we previously grew after a successful shrink and now failed, |
| 82 | + // we've bracketed the node's true limit. Mark calibrated. |
| 83 | + if succeededSinceShrink { |
| 84 | + p.filterLogsRangeCalibrated.Store(true) |
| 85 | + } |
| 86 | + succeededSinceShrink = false |
| 87 | + |
| 88 | + // Multiplicative decrease: shrink by /1.5 |
| 89 | + batchRange = uint64(float64(batchRange) / 1.5) |
| 90 | + if batchRange < 1 { |
| 91 | + batchRange = 1 |
| 92 | + } |
| 93 | + |
| 94 | + continue // retry the same cursor with a smaller range |
| 95 | + } |
| 96 | + |
| 97 | + allLogs = append(allLogs, logs...) |
| 98 | + succeededSinceShrink = true |
| 99 | + |
| 100 | + // Store what worked |
| 101 | + p.filterLogsLastRange.Store(int64(batchRange)) |
| 102 | + |
| 103 | + // Advance past the successfully fetched range |
| 104 | + cursor = end + 1 |
| 105 | + |
| 106 | + // Check context before continuing to next batch |
| 107 | + if err := ctx.Err(); err != nil { |
| 108 | + return nil, err |
| 109 | + } |
| 110 | + |
| 111 | + // Additive increase: try to grow the batch range if not yet calibrated |
| 112 | + if cursor <= to && !p.filterLogsRangeCalibrated.Load() { |
| 113 | + ceiling := p.filterLogsCeiling() |
| 114 | + grown := batchRange + additiveFactor |
| 115 | + if grown > ceiling { |
| 116 | + grown = ceiling |
| 117 | + } |
| 118 | + batchRange = grown |
| 119 | + } |
| 120 | + } |
| 121 | + |
| 122 | + return allLogs, nil |
| 123 | +} |
| 124 | + |
| 125 | +// resolveFilterBlockRange resolves the actual uint64 from/to block numbers from a |
| 126 | +// FilterQuery, fetching the latest block number if ToBlock is nil. |
| 127 | +func (p *Provider) resolveFilterBlockRange(ctx context.Context, q ethereum.FilterQuery) (uint64, uint64, error) { |
| 128 | + var from uint64 |
| 129 | + if q.FromBlock != nil { |
| 130 | + from = q.FromBlock.Uint64() |
| 131 | + } |
| 132 | + |
| 133 | + if q.ToBlock != nil { |
| 134 | + return from, q.ToBlock.Uint64(), nil |
| 135 | + } |
| 136 | + |
| 137 | + // ToBlock is nil → fetch latest block number |
| 138 | + latest, err := p.BlockNumber(ctx) |
| 139 | + if err != nil { |
| 140 | + return 0, 0, err |
| 141 | + } |
| 142 | + return from, latest, nil |
| 143 | +} |
| 144 | + |
| 145 | +// effectiveFilterLogsBatchRange picks the starting batch range for a filterLogsAutoSplit call. |
| 146 | +func (p *Provider) effectiveFilterLogsBatchRange(totalRange uint64) uint64 { |
| 147 | + // If we have a previously discovered value, use it directly |
| 148 | + if last := p.filterLogsLastRange.Load(); last > 0 { |
| 149 | + v := uint64(last) |
| 150 | + // Respect the explicit ceiling if set |
| 151 | + if p.filterLogsMaxRange > 0 && v > uint64(p.filterLogsMaxRange) { |
| 152 | + v = uint64(p.filterLogsMaxRange) |
| 153 | + } |
| 154 | + if v > totalRange { |
| 155 | + return totalRange |
| 156 | + } |
| 157 | + return v |
| 158 | + } |
| 159 | + |
| 160 | + // Not yet calibrated: use explicit max if set |
| 161 | + if p.filterLogsMaxRange > 0 { |
| 162 | + if uint64(p.filterLogsMaxRange) < totalRange { |
| 163 | + return uint64(p.filterLogsMaxRange) |
| 164 | + } |
| 165 | + return totalRange |
| 166 | + } |
| 167 | + |
| 168 | + // Auto mode (filterLogsMaxRange == 0), not yet calibrated: try the full range |
| 169 | + return totalRange |
| 170 | +} |
| 171 | + |
| 172 | +// filterLogsCeiling returns the upper bound for additive increase. |
| 173 | +func (p *Provider) filterLogsCeiling() uint64 { |
| 174 | + if p.filterLogsMaxRange > 0 { |
| 175 | + return uint64(p.filterLogsMaxRange) |
| 176 | + } |
| 177 | + // Auto mode: no artificial ceiling, let AIMD discover it |
| 178 | + return math.MaxUint64 |
| 179 | +} |
| 180 | + |
| 181 | +// isFilterLogsTooMuchDataError checks whether an error from the node indicates |
| 182 | +// the requested range or result set was too large. |
| 183 | +func isFilterLogsTooMuchDataError(err error) bool { |
| 184 | + if err == nil { |
| 185 | + return false |
| 186 | + } |
| 187 | + msg := strings.ToLower(err.Error()) |
| 188 | + for _, pattern := range filterLogsTooMuchDataPatterns { |
| 189 | + if strings.Contains(msg, pattern) { |
| 190 | + return true |
| 191 | + } |
| 192 | + } |
| 193 | + return false |
| 194 | +} |
| 195 | + |
| 196 | +// filterLogsTooMuchDataPatterns are lowercase substrings matched against error messages |
| 197 | +// from various RPC node providers to detect "too much data" / "block range too large" errors. |
| 198 | +var filterLogsTooMuchDataPatterns = []string{ |
| 199 | + "query returned more than", // Infura, Alchemy, generic (e.g. 10000 results) |
| 200 | + "query exceeds max results", // Telos |
| 201 | + "response is too big", // Soneium |
| 202 | + "response exceed size limit", // Various |
| 203 | + "log response size exceeded", // Various |
| 204 | + "block range", // Catches "block range too large", "block range exceeded", etc. |
| 205 | + "too many blocks", // Various |
| 206 | + "logs matched by query exceeds", // Various |
| 207 | + "query timeout exceeded", // Timeout due to large range |
| 208 | + "read limit exceeded", // Various |
| 209 | + "exceed maximum block range", // Various |
| 210 | + "too much data", // Generic |
| 211 | + "result too large", // Generic |
| 212 | +} |
0 commit comments