Method tryQueryUpstreams

erpc/query_executor.go:209–256

(
	ctx context.Context,
	method string,
	attempt func(common.Upstream) error,
)

}

func (qe *EvmQueryExecutor) tryQueryUpstreams(
	ctx context.Context,
	method string,
	attempt func(common.Upstream) error,
) (handled bool, err error) {
	var upstreams []common.Upstream
	if qe.network.policyEngine != nil {
		// Query-shim requests aren't bound to a single finality bucket
		// (a range covers both finalized and unfinalized blocks), so
		// route them through the wildcard slot. Networks configured
		// with `EvalPerFinality: true` should set explicit per-method
		// policies if they want finality-aware routing for queries.
		upstreams = qe.network.policyEngine.GetOrdered(qe.network.Id(), method, "*")
	}
	if len(upstreams) == 0 {
		// Cold-start fallback: engine hasn't published a cache yet.
		for _, u := range qe.network.upstreamsRegistry.GetNetworkUpstreams(ctx, qe.network.Id()) {
			upstreams = append(upstreams, u)
		}
	}
	if len(upstreams) == 0 {
		qe.logger.Debug().Str("method", method).Msgf("no upstreams available for query method")
		return false, nil
	}

	for _, ups := range upstreams {
		if !qe.supportsQueryMethods(ups) {
			qe.logger.Trace().Str("upstreamId", ups.Id()).Str("method", method).Msgf("upstream does not support query streaming, skipping")
			continue
		}

		qe.logger.Debug().Str("upstreamId", ups.Id()).Str("method", method).Msgf("attempting native query pipe-through to upstream")

		err := attempt(ups)
		if err == nil {
			qe.logger.Debug().Str("upstreamId", ups.Id()).Str("method", method).Msgf("native query pipe-through succeeded")
			return true, nil
		}
		if qe.canRetryQueryStream(err) {
			qe.logger.Debug().Err(err).Str("upstreamId", ups.Id()).Str("method", method).Msgf("query pipe-through failed before page emission, trying next upstream")
			continue
		}
		qe.logger.Debug().Err(err).Str("upstreamId", ups.Id()).Str("method", method).Msgf("query pipe-through failed after page emission, cannot retry")
		return true, err
	}

	return false, nil
}

func (qe *EvmQueryExecutor) canRetryQueryStream(err error) bool {
	if err == nil {