Add outcome for previous round missing request ids

.golangci.yml

@@ -6,7 +6,7 @@ linters:
     - depguard
     - errorlint
     - exhaustive
-    - gomodguard
+    - gomodguard_v2
     - gosec
     - loggercheck
     - misspell
@@ -88,9 +88,8 @@ linters:
       errorf: false
     exhaustive:
       default-signifies-exhaustive: true
-    gomodguard:
-      blocked:
-        local-replace-directives: false
+    gomodguard_v2:
+      local-replace-directives: false
     gosec:
       excludes:
         - G101

libs/chainconsensus/oracle/reporting_plugin.go

@@ -27,6 +27,8 @@ import (
 	ctypes "github.com/smartcontractkit/capabilities/libs/chainconsensus/types"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings"
+	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 )
 
 const (
@@ -63,11 +65,12 @@ func (c Config) matchingThreshold() int {
 }
 
 type reportingPlugin struct {
-	config         Config
-	logger         logger.SugaredLogger
-	blocksProvider BlocksProvider
-	requestsStore  RequestsHandler
-	metrics        metrics.ConsensusMetrics
+	config                       Config
+	logger                       logger.SugaredLogger
+	blocksProvider               BlocksProvider
+	requestsStore                RequestsHandler
+	metrics                      metrics.ConsensusMetrics
+	enableMissingRequestRecovery limits.GateLimiter
 }
 
 func newReportingPlugin(
@@ -76,14 +79,22 @@ func newReportingPlugin(
 	blocksProvider BlocksProvider,
 	requestsStore RequestsHandler,
 	metrics metrics.ConsensusMetrics,
-) *reportingPlugin {
-	return &reportingPlugin{
+	limitsFactory limits.Factory,
+) (*reportingPlugin, error) {
+	rp := &reportingPlugin{
 		config:         config,
 		logger:         logger,
 		blocksProvider: blocksProvider,
 		requestsStore:  requestsStore,
 		metrics:        metrics,
 	}
+
+	var err error
+	rp.enableMissingRequestRecovery, err = limits.MakeGateLimiter(limitsFactory, cresettings.Default.MissingRequestRecoveryEnabled)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create missing request recovery gate limiter: %w", err)
+	}
+	return rp, nil
 }
 
 func (rp *reportingPlugin) Query(ctx context.Context, outctx ocr3types.OutcomeContext) (types.Query, error) {
@@ -185,6 +196,12 @@ func (rp *reportingPlugin) Observation(
 		return nil, fmt.Errorf("failed to add observations for missing requests from previous outcome: %w", err)
 	}
 
+	open, err := rp.enableMissingRequestRecovery.IsOpen(ctx)
+	if err != nil {
+		rp.logger.Errorw("error checking if enableMissingRequestRecovery is enabled", "error", err)
+	}
+	observation.EnableMissingRequestRecovery = open
+
 	// add observations for requests provided by the leader in the query
 	err = rp.addObservations(ctx, query.RequestIDs, observation)
 	if err != nil {
@@ -294,6 +311,11 @@ func (rp *reportingPlugin) getMissingRequestIDs(roundRequests map[string]struct{
 		}
 	}
 
+	if len(missingRequestIDs) > 0 {
+		rp.logger.Infow("Proposing missing request IDs: present in local store but absent from the leader's query",
+			"missingRequestIDs", missingRequestIDs)
+	}
+
 	return missingRequestIDs, nil
 }
 
@@ -563,9 +585,34 @@ func (rp *reportingPlugin) agreeOnMissingRequestIDs(aos []attributedObservation)
 	}
 
 	sort.Strings(result)
+	if len(result) > 0 {
+		rp.logger.Infow("Quorum agreed on missing request IDs: committing them to the outcome for recovery in the next round",
+			"missingRequestIDs", result)
+	}
 	return result, nil
 }
 
+// agreeOnFeatureEnableMissingRequestRecovery reaches DON-wide quorum, once per round, on
+// whether missing-request recovery is enabled. This is a per-node, per-round capability flag
+// unrelated to any specific request ID, so it is voted on independently of per-request
+// observation quorum (which the aggregation loop in Outcome enforces on its own).
+// TODO: clean up settings flag and logic when all nodes in all envs support missing-request recovery: https://smartcontract-it.atlassian.net/browse/PLEX-3461
+func (rp *reportingPlugin) agreeOnEnableMissingRequestRecovery(aos []attributedObservation) bool {
+	minMatching := byzQuorumSize(rp.config.N, rp.config.F)
+	counter := 0
+	for _, ob := range aos {
+		if ob.Observation.EnableMissingRequestRecovery {
+			counter++
+			if counter >= minMatching {
+				rp.logger.Infow("Quorum reached for enabling missing request recovery: recovered requests will be aggregated into this round's outcome",
+					"votes", counter, "required", minMatching)
+				return true
+			}
+		}
+	}
+	return false
+}
+
 func (rp *reportingPlugin) agreeOnEventuallyConsistentValue(requestID string, aos []attributedObservation) ([]byte, int, error) {
 	iterator := func(yield func(commontypes.OracleID, *observation[[32]byte, []byte]) bool) {
 		for _, ob := range aos {
@@ -786,7 +833,34 @@ func (rp *reportingPlugin) Outcome(
 		return nil, fmt.Errorf("failed to unmarshal request IDs: %w", err)
 	}
 
-	for _, requestID := range query.RequestIDs {
+	requestIDs := query.RequestIDs
+	if prevOutcome := rp.tryUnmarshalPreviousOutcome(outctx); prevOutcome != nil && rp.agreeOnEnableMissingRequestRecovery(aos) {
+		seen := make(map[string]struct{}, len(query.RequestIDs))
+		for _, requestID := range query.RequestIDs {
+			seen[requestID] = struct{}{}
+		}
+
+		// Requests that the leader's query omitted but that a quorum of nodes still supplied
+		// observations for (via addObservationsOfPrevMissingRequests) must still be aggregated here,
+		// otherwise they would be recycled into MissingRequestIDs forever instead of getting resolved.
+		recoveredRequestIDs := make([]string, 0, len(prevOutcome.MissingRequestIDs))
+		for _, requestID := range prevOutcome.MissingRequestIDs {
+			if _, ok := seen[requestID]; ok {
+				continue
+			}
+			seen[requestID] = struct{}{}
+			requestIDs = append(requestIDs, requestID)
+			recoveredRequestIDs = append(recoveredRequestIDs, requestID)
+		}
+
+		if len(recoveredRequestIDs) > 0 {
+			rp.logger.Infow("Missing request recovery: adding previously missing requests to this round's outcome "+
+				"(they were omitted by the leader's query but agreed upon by a quorum in the previous round)",
+				"recoveredRequestIDs", recoveredRequestIDs)
+		}
+	}
+
+	for _, requestID := range requestIDs {
 		observationType, err := rp.agreeOnObservationType(requestID, aos)
 		if err != nil {
 			rp.logger.Infow("Could not determine observation type", "requestID", requestID, "err", err)
@@ -969,5 +1043,5 @@ func (rp *reportingPlugin) ShouldTransmitAcceptedReport(
 }
 
 func (rp *reportingPlugin) Close() error {
-	return nil
+	return rp.enableMissingRequestRecovery.Close()
 }

libs/chainconsensus/oracle/reporting_plugin_factory.go

@@ -8,6 +8,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 
 	"github.com/smartcontractkit/capabilities/libs/chainconsensus/metrics"
 
@@ -21,19 +22,22 @@ type ReportingPluginFactory struct {
 	requestsStore  RequestsHandler
 	blocksProvider BlocksProvider
 	metrics        metrics.ConsensusMetrics
+	limitsFactory  limits.Factory
 }
 
 func NewReportingPluginFactory(
 	logger logger.SugaredLogger,
 	requestsStore RequestsHandler,
 	blocksProvider BlocksProvider,
 	metrics metrics.ConsensusMetrics,
+	limitsFactory limits.Factory,
 ) *ReportingPluginFactory {
 	return &ReportingPluginFactory{
 		logger:         logger,
 		requestsStore:  requestsStore,
 		blocksProvider: blocksProvider,
 		metrics:        metrics,
+		limitsFactory:  limitsFactory,
 	}
 }
 
@@ -69,7 +73,12 @@ func (rpf *ReportingPluginFactory) NewReportingPlugin(
 
 	pluginLogger := rpf.logger.Named("ChainReadReportingPlugin")
 
-	return newReportingPlugin(cfg, pluginLogger, rpf.blocksProvider, rpf.requestsStore, rpf.metrics), ocr3types.ReportingPluginInfo{
+	plugin, err := newReportingPlugin(cfg, pluginLogger, rpf.blocksProvider, rpf.requestsStore, rpf.metrics, rpf.limitsFactory)
+	if err != nil {
+		return nil, ocr3types.ReportingPluginInfo{}, fmt.Errorf("failed to create reporting plugin: %w", err)
+	}
+
+	return plugin, ocr3types.ReportingPluginInfo{
 		Name: rpf.logger.Name() + ".chain-reads-oracle",
 		Limits: ocr3types.ReportingPluginLimits{
 			MaxQueryLength:       int(offchainCfg.MaxQueryLengthBytes),