Method Run

consensus/executor.go:118–166

Run is the main entry point for the consensus executor. It delegates to executeConsensus which decouples caller-visible latency from analysis completion (see runAnalyzer).

(
	ctx context.Context,
	originalReq *common.NormalizedRequest,
	in inner,
)

// It delegates to executeConsensus which decouples caller-visible
// latency from analysis completion (see runAnalyzer).
func (e *executor) Run(
	ctx context.Context,
	originalReq *common.NormalizedRequest,
	in inner,
) (*common.NormalizedResponse, error) {
	startTime := time.Now()

	if originalReq == nil {
		e.logger.Error().Msg("Unexpected nil request in consensus policy")
		return in(ctx, originalReq)
	}

	// Tag-based participant quota (opt-in). Front-load enough tag-matching
	// upstreams so the first maxParticipants drawn by the slots below
	// include the configured minimum from each required group. Runs at the
	// very top of consensus, before any participant slot consumes an
	// upstream (req.UpstreamIdx is still 0), so the reorder takes effect.
	// Best-effort: shortfalls fall through to lowParticipantsBehavior /
	// agreementThreshold like organic low participation.
	if len(e.config.requiredParticipants) > 0 {
		if reordered := reorderForParticipantQuota(originalReq.Upstreams(), e.config.requiredParticipants); len(reordered) > 0 {
			originalReq.SetUpstreams(reordered)
		}
	}

	labels := e.extractMetricsLabels(ctx, originalReq)
	ctx, consensusSpan := e.startConsensusSpan(ctx, labels)
	defer consensusSpan.End()

	lg := e.logger.With().
		Interface("id", originalReq.ID()).
		Str("component", "consensus").
		Str("networkId", labels.networkId).
		Logger()

	out := e.executeConsensus(
		ctx,
		&lg,
		originalReq,
		labels,
		in,
		startTime,
		consensusSpan,
	)
	if out == nil {
		return nil, nil
	}
	return out.Result, out.Error
}

// executeConsensus decouples two distinct concerns:
//