executeParticipant runs a single upstream request within a goroutine.
( ctx context.Context, lg *zerolog.Logger, labels metricsLabels, in inner, req *common.NormalizedRequest, index int, responseChan chan<- *execResult, )
| 722 | |
| 723 | // executeParticipant runs a single upstream request within a goroutine. |
| 724 | func (e *executor) executeParticipant( |
| 725 | ctx context.Context, |
| 726 | lg *zerolog.Logger, |
| 727 | labels metricsLabels, |
| 728 | in inner, |
| 729 | req *common.NormalizedRequest, |
| 730 | index int, |
| 731 | responseChan chan<- *execResult, |
| 732 | ) { |
| 733 | // Panic recovery |
| 734 | defer func() { |
| 735 | if r := recover(); r != nil { |
| 736 | lg.Error(). |
| 737 | Interface("panic", r). |
| 738 | Int("index", index). |
| 739 | Str("stack", string(debug.Stack())). |
| 740 | Msg("Panic in consensus participant") |
| 741 | telemetry.MetricConsensusPanics.WithLabelValues(labels.projectId, labels.networkId, labels.category, labels.finalityStr).Inc() |
| 742 | responseChan <- &execResult{Err: errPanicInConsensus} |
| 743 | } |
| 744 | }() |
| 745 | |
| 746 | // Check for cancellation before execution |
| 747 | if ctx.Err() != nil { |
| 748 | telemetry.MetricConsensusCancellations. |
| 749 | WithLabelValues(labels.projectId, labels.networkId, labels.category, "before_execution", labels.finalityStr). |
| 750 | Inc() |
| 751 | responseChan <- nil |
| 752 | return |
| 753 | } |
| 754 | |
| 755 | // Execute the slot inner — returns (response, error) directly. |
| 756 | respObj, respErr := in(ctx, req) |
| 757 | |
| 758 | // Track post-execution cancellations for observability, but do NOT discard |
| 759 | // the result. The result is still valid and should participate in |
| 760 | // consensus analysis. |
| 761 | if ctx.Err() != nil { |
| 762 | telemetry.MetricConsensusCancellations. |
| 763 | WithLabelValues(labels.projectId, labels.networkId, labels.category, "after_execution", labels.finalityStr). |
| 764 | Inc() |
| 765 | } |
| 766 | |
| 767 | if respObj == nil && respErr == nil { |
| 768 | responseChan <- nil |
| 769 | return |
| 770 | } |
| 771 | |
| 772 | var upstream common.Upstream |
| 773 | if respObj != nil { |
| 774 | upstream = respObj.Upstream() |
| 775 | } |
| 776 | if upstream == nil && respErr != nil { |
| 777 | var uae interface{ Upstream() common.Upstream } |
| 778 | if errors.As(respErr, &uae) { |
| 779 | upstream = uae.Upstream() |
| 780 | } |
| 781 | var uxe *common.ErrUpstreamsExhausted |