[DO NOT MERGE] Test combine solcap read trigger and min identical responses changes

core/capabilities/launcher.go

@@ -493,7 +493,7 @@
 	}
 }
 
 func (w *launcher) addRemoteCapability(ctx context.Context, cid string, capabilityConfig capabilities.CapabilityConfiguration, myDON registrysyncer.DON, remoteDON registrysyncer.DON, localRegistry *registrysyncer.LocalRegistry) error {
 	capability, ok := localRegistry.IDsToCapabilities[cid]
 	if !ok {
 		return fmt.Errorf("could not find capability matching id %s", cid)
@@ -592,7 +592,7 @@
 				w.cachedShims.executableClients[shimKey] = execCap
 			}
 			// V1 capabilities read transmission schedule from every request
-			if errCfg := execCap.SetConfig(info, myDON.DON, defaultTargetRequestTimeout, nil, nil); errCfg != nil {
+			if errCfg := execCap.SetConfig(info, myDON.DON, defaultTargetRequestTimeout, nil, nil, 0); errCfg != nil {
 				return nil, fmt.Errorf("failed to set trigger config: %w", errCfg)
 			}
 			return execCap.(capabilityService), nil
@@ -618,7 +618,7 @@
 				w.cachedShims.executableClients[shimKey] = execCap
 			}
 			// V1 capabilities read transmission schedule from every request
-			if errCfg := execCap.SetConfig(info, myDON.DON, defaultTargetRequestTimeout, nil, nil); errCfg != nil {
+			if errCfg := execCap.SetConfig(info, myDON.DON, defaultTargetRequestTimeout, nil, nil, 0); errCfg != nil {
 				return nil, fmt.Errorf("failed to set trigger config: %w", errCfg)
 			}
 			return execCap.(capabilityService), nil
@@ -989,7 +989,7 @@
 			if err != nil {
 				return fmt.Errorf("failed to get signers for executable client: %w", err)
 			}
-			err = client.SetConfig(info, myDON.DON, config.RemoteExecutableConfig.RequestTimeout, transmissionConfig, signers)
+			err = client.SetConfig(info, myDON.DON, config.RemoteExecutableConfig.RequestTimeout, transmissionConfig, signers, config.RemoteExecutableConfig.MinResponsesToAggregate)
 			if err != nil {
 				w.lggr.Errorw("failed to update client config", "capID", capID, "method", method, "error", err)
 				continue

core/capabilities/remote/executable/client.go

@@ -48,12 +48,15 @@ type dynamicConfig struct {
 	transmissionConfig *transmission.TransmissionConfig
 	// Has to be set only for V2 capabilities using OCR.
 	signers [][]byte
+	// minResponsesToAggregate overrides the default F+1 matching threshold for read consensus.
+	// 0 defaults to F+1.
+	minResponsesToAggregate uint32
 }
 
 type Client interface {
 	commoncap.ExecutableCapability
 	Receive(ctx context.Context, msg *types.MessageBody)
-	SetConfig(remoteCapabilityInfo commoncap.CapabilityInfo, localDonInfo commoncap.DON, requestTimeout time.Duration, transmissionConfig *transmission.TransmissionConfig, signers [][]byte) error
+	SetConfig(remoteCapabilityInfo commoncap.CapabilityInfo, localDonInfo commoncap.DON, requestTimeout time.Duration, transmissionConfig *transmission.TransmissionConfig, signers [][]byte, minResponsesToAggregate uint32) error
 }
 
 var _ Client = &client{}
@@ -80,7 +83,7 @@ func NewClient(capabilityID string, capMethodName string, dispatcher types.Dispa
 
 // SetConfig sets the remote capability configuration dynamically
 // TransmissionConfig has to be set only for V2 capabilities. V1 capabilities read transmission schedule from every request.
-func (c *client) SetConfig(remoteCapabilityInfo commoncap.CapabilityInfo, localDonInfo commoncap.DON, requestTimeout time.Duration, transmissionConfig *transmission.TransmissionConfig, signers [][]byte) error {
+func (c *client) SetConfig(remoteCapabilityInfo commoncap.CapabilityInfo, localDonInfo commoncap.DON, requestTimeout time.Duration, transmissionConfig *transmission.TransmissionConfig, signers [][]byte, minResponsesToAggregate uint32) error {
 	if remoteCapabilityInfo.ID == "" || remoteCapabilityInfo.ID != c.capabilityID {
 		return fmt.Errorf("capability info provided does not match the client's capabilityID: %s != %s", remoteCapabilityInfo.ID, c.capabilityID)
 	}
@@ -96,11 +99,12 @@ func (c *client) SetConfig(remoteCapabilityInfo commoncap.CapabilityInfo, localD
 
 	// always replace the whole dynamicConfig object to avoid inconsistent state
 	c.cfg.Store(&dynamicConfig{
-		remoteCapabilityInfo: remoteCapabilityInfo,
-		localDONInfo:         localDonInfo,
-		requestTimeout:       requestTimeout,
-		transmissionConfig:   transmissionConfig,
-		signers:              signers,
+		remoteCapabilityInfo:    remoteCapabilityInfo,
+		localDONInfo:            localDonInfo,
+		requestTimeout:          requestTimeout,
+		transmissionConfig:      transmissionConfig,
+		signers:                 signers,
+		minResponsesToAggregate: minResponsesToAggregate,
 	})
 	c.lggr.Infow("SetConfig", "remoteDONName", remoteCapabilityInfo.DON.Name, "remoteDONID", remoteCapabilityInfo.DON.ID, "requestTimeout", requestTimeout, "transmissionConfig", transmissionConfig)
 	return nil
@@ -233,7 +237,7 @@ func (c *client) Execute(ctx context.Context, capReq commoncap.CapabilityRequest
 	}
 
 	req, err := request.NewClientExecuteRequest(ctx, c.lggr, capReq, cfg.remoteCapabilityInfo, cfg.localDONInfo, c.dispatcher,
-		cfg.requestTimeout, cfg.transmissionConfig, c.capMethodName, cfg.signers)
+		cfg.requestTimeout, cfg.transmissionConfig, c.capMethodName, cfg.signers, cfg.minResponsesToAggregate)
 	if err != nil {
 		return commoncap.CapabilityResponse{}, fmt.Errorf("failed to create client request: %w", err)
 	}

core/capabilities/remote/executable/client_test.go

@@ -338,7 +338,7 @@ func testClient(t *testing.T, numWorkflowPeers int, workflowNodeResponseTimeout
 	for i := range numWorkflowPeers {
 		workflowPeerDispatcher := broker.NewDispatcherForNode(workflowPeers[i])
 		caller := executable.NewClient(capInfo.ID, "", workflowPeerDispatcher, lggr)
-		err := caller.SetConfig(capInfo, workflowDonInfo, workflowNodeResponseTimeout, nil, nil)
+		err := caller.SetConfig(capInfo, workflowDonInfo, workflowNodeResponseTimeout, nil, nil, 0)
 		require.NoError(t, err)
 		servicetest.Run(t, caller)
 		broker.RegisterReceiverNode(workflowPeers[i], caller)
@@ -468,7 +468,7 @@ func (t *clientTestServer) sendResponse(messageID string, responseErr error,
 
 type clientSetConfigTestFixture struct {
 	Client interface {
-		SetConfig(commoncap.CapabilityInfo, commoncap.DON, time.Duration, *transmission.TransmissionConfig, [][]byte) error
+		SetConfig(commoncap.CapabilityInfo, commoncap.DON, time.Duration, *transmission.TransmissionConfig, [][]byte, uint32) error
 		Info(context.Context) (commoncap.CapabilityInfo, error)
 		Start(context.Context) error
 		Close() error
@@ -523,7 +523,7 @@ func TestClient_SetConfig(t *testing.T) {
 			DeltaStage: 10 * time.Millisecond,
 		}
 
-		err := fixture.Client.SetConfig(fixture.ValidCapInfo, fixture.ValidDonInfo, fixture.ValidTimeout, transmissionConfig, nil)
+		err := fixture.Client.SetConfig(fixture.ValidCapInfo, fixture.ValidDonInfo, fixture.ValidTimeout, transmissionConfig, nil, 0)
 		require.NoError(t, err)
 
 		info, err := fixture.Client.Info(t.Context())
@@ -541,7 +541,7 @@ func TestClient_SetConfig(t *testing.T) {
 			CapabilityType: commoncap.CapabilityTypeAction,
 		}
 
-		err := fixture.Client.SetConfig(invalidCapInfo, fixture.ValidDonInfo, fixture.ValidTimeout, nil, nil)
+		err := fixture.Client.SetConfig(invalidCapInfo, fixture.ValidDonInfo, fixture.ValidTimeout, nil, nil, 0)
 		require.Error(t, err)
 		assert.Contains(t, err.Error(), "capability info provided does not match the client's capabilityID")
 		assert.Contains(t, err.Error(), "different_capability@1.0.0 != test_capability@1.0.0")
@@ -558,7 +558,7 @@ func TestClient_SetConfig(t *testing.T) {
 			F:       0,
 		}
 
-		err := fixture.Client.SetConfig(fixture.ValidCapInfo, invalidDonInfo, fixture.ValidTimeout, nil, nil)
+		err := fixture.Client.SetConfig(fixture.ValidCapInfo, invalidDonInfo, fixture.ValidTimeout, nil, nil, 0)
 		require.Error(t, err)
 		assert.Contains(t, err.Error(), "empty localDonInfo provided")
 	})
@@ -569,7 +569,7 @@ func TestClient_SetConfig(t *testing.T) {
 		fixture := newClientSetConfigTestFixture(t)
 
 		initialTimeout := 10 * time.Second
-		err := fixture.Client.SetConfig(fixture.ValidCapInfo, fixture.ValidDonInfo, initialTimeout, nil, nil)
+		err := fixture.Client.SetConfig(fixture.ValidCapInfo, fixture.ValidDonInfo, initialTimeout, nil, nil, 0)
 		require.NoError(t, err)
 
 		newTimeout := 60 * time.Second
@@ -579,7 +579,7 @@ func TestClient_SetConfig(t *testing.T) {
 			F:       1,
 		}
 
-		err = fixture.Client.SetConfig(fixture.ValidCapInfo, newDonInfo, newTimeout, nil, nil)
+		err = fixture.Client.SetConfig(fixture.ValidCapInfo, newDonInfo, newTimeout, nil, nil, 0)
 		require.NoError(t, err)
 
 		info, err := fixture.Client.Info(t.Context())
@@ -607,7 +607,7 @@ func TestClient_SetConfig_StartClose(t *testing.T) {
 		fixture := newClientSetConfigTestFixture(t)
 		ctx := t.Context()
 
-		require.NoError(t, fixture.Client.SetConfig(fixture.ValidCapInfo, fixture.ValidDonInfo, fixture.ValidTimeout, nil, nil))
+		require.NoError(t, fixture.Client.SetConfig(fixture.ValidCapInfo, fixture.ValidDonInfo, fixture.ValidTimeout, nil, nil, 0))
 		require.NoError(t, fixture.Client.Start(ctx))
 		require.NoError(t, fixture.Client.Close())
 	})
@@ -618,12 +618,12 @@ func TestClient_SetConfig_StartClose(t *testing.T) {
 		fixture := newClientSetConfigTestFixture(t)
 		ctx := t.Context()
 
-		require.NoError(t, fixture.Client.SetConfig(fixture.ValidCapInfo, fixture.ValidDonInfo, fixture.ValidTimeout, nil, nil))
+		require.NoError(t, fixture.Client.SetConfig(fixture.ValidCapInfo, fixture.ValidDonInfo, fixture.ValidTimeout, nil, nil, 0))
 		require.NoError(t, fixture.Client.Start(ctx))
 
 		newCapInfo := fixture.ValidCapInfo
 		newCapInfo.Description = "new description"
-		require.NoError(t, fixture.Client.SetConfig(newCapInfo, fixture.ValidDonInfo, fixture.ValidTimeout, nil, nil))
+		require.NoError(t, fixture.Client.SetConfig(newCapInfo, fixture.ValidDonInfo, fixture.ValidTimeout, nil, nil, 0))
 
 		info, err := fixture.Client.Info(ctx)
 		require.NoError(t, err)

core/capabilities/remote/executable/endtoend_test.go

@@ -283,7 +283,7 @@ func setupRemoteExecutableHarness(t *testing.T, underlying commoncap.ExecutableC
 	for i := range numWorkflowPeers {
 		workflowPeerDispatcher := broker.NewDispatcherForNode(workflowPeers[i])
 		workflowNode := executable.NewClient(capInfo.ID, "", workflowPeerDispatcher, lggr)
-		err := workflowNode.SetConfig(capInfo, workflowDonInfo, workflowNodeTimeout, nil, nil)
+		err := workflowNode.SetConfig(capInfo, workflowDonInfo, workflowNodeTimeout, nil, nil, 0)
 		require.NoError(t, err)
 		servicetest.Run(t, workflowNode)
 		broker.RegisterReceiverNode(workflowPeers[i], workflowNode)

core/capabilities/remote/executable/request/client_request.go

@@ -67,7 +67,7 @@
 }
 
 // ErrResponseQuorumUnreachable is returned when enough peer responses have been
-// received that F+1 matching payloads are no longer mathematically possible.
+// received that the required number of matching payloads (configurable; defaults to F+1) are no longer mathematically possible.
 var ErrResponseQuorumUnreachable = errors.New("response quorum unreachable: not enough matching capability responses")
 
 type clientResponse struct {
@@ -102,10 +102,10 @@
 }
 
 // TransmissionConfig has to be set only for V2 capabilities. V1 capabilities read transmission schedule from every request.
 func NewClientExecuteRequest(ctx context.Context, lggr logger.Logger, req commoncap.CapabilityRequest,
 	remoteCapabilityInfo commoncap.CapabilityInfo, localDonInfo commoncap.DON, dispatcher types.Dispatcher,
 	requestTimeout time.Duration, transmissionConfig *transmission.TransmissionConfig, capMethodName string,
-	signers [][]byte,
+	signers [][]byte, minResponsesToAggregate uint32,
 ) (*ClientRequest, error) {
 	rawRequest, err := proto.MarshalOptions{Deterministic: true}.Marshal(pb.CapabilityRequestToProto(req))
 	if err != nil {
@@ -135,7 +135,7 @@
 	}
 
 	lggr = logger.With(lggr, "requestId", requestID) // cap ID and method name included in the parent logger
-	return newClientRequest(ctx, lggr, requestID, remoteCapabilityInfo, localDonInfo, dispatcher, requestTimeout, tc, types.MethodExecute, rawRequest, workflowExecutionID, req.Metadata.ReferenceID, capMethodName, signers)
+	return newClientRequest(ctx, lggr, requestID, remoteCapabilityInfo, localDonInfo, dispatcher, requestTimeout, tc, types.MethodExecute, rawRequest, workflowExecutionID, req.Metadata.ReferenceID, capMethodName, signers, minResponsesToAggregate)
 }
 
 var (
@@ -145,10 +145,20 @@
 	defaultResponseAggregationGrace = 10 * time.Second
 )
 
+// requiredConfirmations returns the minimum number of matching peer responses needed before
+// the workflow DON accepts a read result. When minResponsesToAggregate is non-zero it is used
+// directly; otherwise the default of F+1 is used (backward compatible).
+func requiredConfirmations(f uint8, minResponsesToAggregate uint32) int {
+	if minResponsesToAggregate > 0 {
+		return int(minResponsesToAggregate)
+	}
+	return int(f) + 1
+}
+
 func newClientRequest(ctx context.Context, lggr logger.Logger, requestID string, remoteCapabilityInfo commoncap.CapabilityInfo,
 	localDonInfo commoncap.DON, dispatcher types.Dispatcher, requestTimeout time.Duration,
 	tc transmission.TransmissionConfig, methodType string, rawRequest []byte, workflowExecutionID string, stepRef string, capMethodName string,
-	signers [][]byte,
+	signers [][]byte, minResponsesToAggregate uint32,
 ) (*ClientRequest, error) {
 	remoteCapabilityDonInfo := remoteCapabilityInfo.DON
 	if remoteCapabilityDonInfo == nil {
@@ -248,7 +258,7 @@
 		cancelFn:                      cancelFn,
 		createdAt:                     time.Now(),
 		requestTimeout:                requestTimeout,
-		requiredResponseConfirmations: int(remoteCapabilityDonInfo.F + 1),
+		requiredResponseConfirmations: requiredConfirmations(remoteCapabilityDonInfo.F, minResponsesToAggregate),
 		remoteNodeCount:               len(remoteCapabilityDonInfo.Members),
 		responseIDCount:               make(map[[32]byte]int),
 		meteringResponses:             make(map[[32]byte][]commoncap.MeteringNodeDetail),