remote request: extract chain-specific response handling into pluggable policy and move Aptos failed-hash logic out of ClientRequest

Author: cawthorne

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

@@ -7,7 +7,6 @@ import (
 	"errors"
 	"fmt"
 	"sort"
-	"strings"
 	"sync"
 	"time"
 
@@ -18,7 +17,6 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb"
-	aptoscap "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/aptos"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-protos/workflows/go/events"
 
@@ -48,17 +46,10 @@ type ClientRequest struct {
 
 	requiredIdenticalResponses int
 	remoteNodeCount            int
-	failedHashQuorumResponses  int
-	capabilityID               string
-	capabilityMethod           string
 
 	requestTimeout time.Duration
 
-	aptosFailedHashCount    map[string]int
-	aptosFailedHashPayload  map[string][]byte
-	aptosFailedHashMetering map[string][]commoncap.MeteringNodeDetail
-	aptosFailedTotal        int
-	aptosWriteSuccessSeen   bool
+	responsePolicy responsePolicy
 
 	respSent bool
 	mux      sync.Mutex
@@ -194,28 +185,18 @@ func newClientRequest(ctx context.Context, lggr logger.Logger, requestID string,
 		}(ctxWithCancel, peerID, delay)
 	}
 
-	failedHashQuorumResponses := min(len(remoteCapabilityDonInfo.Members), int(2*remoteCapabilityDonInfo.F+1))
-	if failedHashQuorumResponses < 1 {
-		failedHashQuorumResponses = 1
-	}
-
 	return &ClientRequest{
 		id:                         requestID,
 		cancelFn:                   cancelFn,
 		createdAt:                  time.Now(),
 		requestTimeout:             requestTimeout,
 		requiredIdenticalResponses: int(remoteCapabilityDonInfo.F + 1),
 		remoteNodeCount:            len(remoteCapabilityDonInfo.Members),
-		failedHashQuorumResponses:  failedHashQuorumResponses,
-		capabilityID:               remoteCapabilityInfo.ID,
-		capabilityMethod:           capMethodName,
 		responseIDCount:            make(map[[32]byte]int),
 		meteringResponses:          make(map[[32]byte][]commoncap.MeteringNodeDetail),
 		errorCount:                 make(map[string]int),
 		responseReceived:           responseReceived,
-		aptosFailedHashCount:       make(map[string]int),
-		aptosFailedHashPayload:     make(map[string][]byte),
-		aptosFailedHashMetering:    make(map[string][]commoncap.MeteringNodeDetail),
+		responsePolicy:             newResponsePolicy(remoteCapabilityInfo, capMethodName),
 		responseCh:                 make(chan clientResponse, 1),
 		wg:                         &wg,
 		lggr:                       lggr,
@@ -285,13 +266,15 @@ func (c *ClientRequest) Cancel(err error) {
 	c.mux.Lock()
 	defer c.mux.Unlock()
 	if !c.respSent {
-		payload, ok, buildErr := c.buildDeterministicAptosFailedResponse(true)
-		if buildErr != nil {
-			c.lggr.Warnw("failed to build deterministic Aptos failed response", "error", buildErr)
-		}
-		if ok {
-			c.sendResponse(clientResponse{Result: payload})
-			return
+		if c.responsePolicy != nil {
+			payload, ok, buildErr := c.responsePolicy.BuildDeterministicResponse(true)
+			if buildErr != nil {
+				c.lggr.Warnw("failed to build deterministic policy response", "error", buildErr)
+			}
+			if ok {
+				c.sendResponse(clientResponse{Result: payload})
+				return
+			}
 		}
 		c.sendResponse(clientResponse{Err: err})
 	}
@@ -370,15 +353,17 @@ func (c *ClientRequest) OnMessage(_ context.Context, msg *types.MessageBody) err
 		}
 
 		if !c.respSent {
-			c.recordAptosFailedResponse(msg, metadata)
-			payload, ok, buildErr := c.buildDeterministicAptosFailedResponse(c.allResponsesReceived())
-			if buildErr != nil {
-				return fmt.Errorf("failed to build deterministic Aptos failed response: %w", buildErr)
-			}
-			if ok {
-				c.sendResponse(clientResponse{Result: payload})
-			} else if err := c.maybeFinalizeAptosWriteAfterAllResponses(); err != nil {
-				return err
+			if c.responsePolicy != nil {
+				c.responsePolicy.ObserveOKResponse(msg, metadata)
+				payload, ok, buildErr := c.responsePolicy.BuildDeterministicResponse(c.allResponsesReceived())
+				if buildErr != nil {
+					return fmt.Errorf("failed to build deterministic policy response: %w", buildErr)
+				}
+				if ok {
+					c.sendResponse(clientResponse{Result: payload})
+				} else if err := c.maybeFinalizeResponsePolicyAfterAllResponses(); err != nil {
+					return err
+				}
 			}
 		}
 	} else {
@@ -390,8 +375,8 @@ func (c *ClientRequest) OnMessage(_ context.Context, msg *types.MessageBody) err
 			c.lggr.Warn("received multiple different errors for the same request, number of different errors received: %d", len(c.errorCount))
 		}
 
-		if c.isAptosWriteReportRequest() {
-			if err := c.maybeFinalizeAptosWriteAfterAllResponses(); err != nil {
+		if c.responsePolicy != nil && c.responsePolicy.ShouldDeferErrorResponses() {
+			if err := c.maybeFinalizeResponsePolicyAfterAllResponses(); err != nil {
 				return err
 			}
 			return nil
@@ -447,25 +432,11 @@ func (c *ClientRequest) encodePayloadWithMetadata(msg *types.MessageBody, metada
 	return pb.MarshalCapabilityResponse(resp)
 }
 
-func (c *ClientRequest) isAptosWriteReportRequest() bool {
-	if c.capabilityMethod != "WriteReport" {
-		return false
-	}
-
-	return strings.HasPrefix(c.capabilityID, "aptos:")
-}
-
 func (c *ClientRequest) shouldDeferIdenticalResponse(payload []byte) bool {
-	if !c.isAptosWriteReportRequest() {
-		return false
-	}
-
-	reply, ok := decodeAptosWriteReportReply(payload)
-	if !ok {
+	if c.responsePolicy == nil {
 		return false
 	}
-
-	return reply.GetTxStatus() == aptoscap.TxStatus_TX_STATUS_FAILED
+	return c.responsePolicy.ShouldDeferIdenticalResponse(payload)
 }
 
 func (c *ClientRequest) allResponsesReceived() bool {
@@ -482,160 +453,23 @@ func (c *ClientRequest) allResponsesReceived() bool {
 	return true
 }
 
-func (c *ClientRequest) recordAptosFailedResponse(msg *types.MessageBody, metadata commoncap.ResponseMetadata) {
-	if !c.isAptosWriteReportRequest() {
-		return
-	}
-
-	reply, ok := decodeAptosWriteReportReply(msg.Payload)
-	if !ok {
-		return
-	}
-	if reply.GetTxStatus() == aptoscap.TxStatus_TX_STATUS_SUCCESS {
-		c.aptosWriteSuccessSeen = true
-		return
-	}
-	if reply.GetTxStatus() != aptoscap.TxStatus_TX_STATUS_FAILED || len(reply.GetTxHash()) == 0 {
-		return
-	}
-
-	normalizedHash, ok := normalizeAptosTxHash(reply.GetTxHash())
-	if !ok {
-		return
-	}
-
-	c.aptosFailedHashCount[normalizedHash]++
-	c.aptosFailedTotal++
-
-	if _, exists := c.aptosFailedHashPayload[normalizedHash]; !exists {
-		c.aptosFailedHashPayload[normalizedHash] = append([]byte(nil), msg.Payload...)
-	}
-
-	if len(metadata.Metering) > 0 {
-		c.aptosFailedHashMetering[normalizedHash] = append(c.aptosFailedHashMetering[normalizedHash], metadata.Metering...)
-	}
-}
-
-func (c *ClientRequest) buildDeterministicAptosFailedResponse(allowNoQuorum bool) ([]byte, bool, error) {
-	if !c.isAptosWriteReportRequest() {
-		return nil, false, nil
-	}
-	if c.aptosWriteSuccessSeen {
-		return nil, false, nil
-	}
-
-	if len(c.aptosFailedHashCount) == 0 {
-		return nil, false, nil
-	}
-	if !allowNoQuorum && c.aptosFailedTotal < c.failedHashQuorumResponses {
-		return nil, false, nil
-	}
-
-	selectedHash := ""
-	for hash := range c.aptosFailedHashPayload {
-		if selectedHash == "" || hash < selectedHash {
-			selectedHash = hash
-		}
-	}
-
-	if selectedHash == "" {
-		return nil, false, nil
-	}
-
-	selectedPayload, ok := c.aptosFailedHashPayload[selectedHash]
-	if !ok || len(selectedPayload) == 0 {
-		return nil, false, nil
-	}
-
-	nodeReports := c.aptosFailedHashMetering[selectedHash]
-	metadata := commoncap.ResponseMetadata{
-		Metering: append([]commoncap.MeteringNodeDetail(nil), nodeReports...),
-	}
-
-	payload, err := buildAptosFailedWriteReportPayload(selectedPayload, selectedHash, metadata)
-	if err != nil {
-		return nil, false, err
-	}
-
-	return payload, true, nil
-}
-
-func decodeAptosWriteReportReply(payload []byte) (*aptoscap.WriteReportReply, bool) {
-	resp, err := pb.UnmarshalCapabilityResponse(payload)
-	if err != nil {
-		return nil, false
-	}
-
-	reply := &aptoscap.WriteReportReply{}
-	if _, err := commoncap.UnwrapResponse(resp, reply); err != nil {
-		return nil, false
-	}
-
-	return reply, true
-}
-
-func buildAptosFailedWriteReportPayload(payload []byte, normalizedHash string, metadata commoncap.ResponseMetadata) ([]byte, error) {
-	resp, err := pb.UnmarshalCapabilityResponse(payload)
-	if err != nil {
-		return nil, err
-	}
-
-	reply := &aptoscap.WriteReportReply{}
-	migrated, err := commoncap.UnwrapResponse(resp, reply)
-	if err != nil {
-		return nil, err
-	}
-
-	reply.TxStatus = aptoscap.TxStatus_TX_STATUS_FAILED
-	reply.TxHash = []byte("0x" + normalizedHash)
-
-	if err := commoncap.SetResponse(&resp, migrated, reply); err != nil {
-		return nil, err
-	}
-
-	resp.Metadata = metadata
-	return pb.MarshalCapabilityResponse(resp)
-}
-
-func normalizeAptosTxHash(raw []byte) (string, bool) {
-	if len(raw) == 32 {
-		return hex.EncodeToString(raw), true
-	}
-
-	s := strings.TrimSpace(strings.ToLower(string(raw)))
-	s = strings.TrimPrefix(s, "0x")
-	if len(s) != 64 {
-		return "", false
-	}
-
-	if _, err := hex.DecodeString(s); err != nil {
-		return "", false
-	}
-
-	return s, true
-}
-
-func (c *ClientRequest) maybeFinalizeAptosWriteAfterAllResponses() error {
-	if !c.isAptosWriteReportRequest() || c.respSent || !c.allResponsesReceived() {
+func (c *ClientRequest) maybeFinalizeResponsePolicyAfterAllResponses() error {
+	if c.responsePolicy == nil || c.respSent {
 		return nil
 	}
 
-	payload, ok, buildErr := c.buildDeterministicAptosFailedResponse(true)
-	if buildErr != nil {
-		return fmt.Errorf("failed to build deterministic Aptos failed response: %w", buildErr)
+	response, ok, err := c.responsePolicy.FinalizeAfterAllResponses(responsePolicyState{
+		AllResponsesReceived: c.allResponsesReceived(),
+		ResponseVariants:     len(c.responseIDCount),
+		TotalErrorCount:      c.totalErrorCount,
+	})
+	if err != nil {
+		return err
 	}
-	if ok {
-		c.sendResponse(clientResponse{Result: payload})
+	if !ok || response == nil {
 		return nil
 	}
 
-	c.sendResponse(clientResponse{
-		Err: fmt.Errorf(
-			"received all Aptos write responses without deterministic failed hash (okResponses=%d errorResponses=%d failedHashResponses=%d)",
-			len(c.responseIDCount),
-			c.totalErrorCount,
-			c.aptosFailedTotal,
-		),
-	})
+	c.sendResponse(*response)
 	return nil
 }

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

@@ -0,0 +1,41 @@
+package request
+
+import (
+	commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+
+	"github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types"
+)
+
+// responsePolicy allows request finalization behavior to be extended for specific
+// capabilities without coupling generic request handling to chain-specific logic.
+type responsePolicy interface {
+	ShouldDeferIdenticalResponse(payload []byte) bool
+	ObserveOKResponse(msg *types.MessageBody, metadata commoncap.ResponseMetadata)
+	BuildDeterministicResponse(allowNoQuorum bool) ([]byte, bool, error)
+	ShouldDeferErrorResponses() bool
+	FinalizeAfterAllResponses(state responsePolicyState) (*clientResponse, bool, error)
+}
+
+type responsePolicyState struct {
+	AllResponsesReceived bool
+	ResponseVariants     int
+	TotalErrorCount      int
+}
+
+type responsePolicyBuilder func(remoteCapabilityInfo commoncap.CapabilityInfo, capabilityMethod string) responsePolicy
+
+var responsePolicyBuilders []responsePolicyBuilder
+
+func registerResponsePolicyBuilder(builder responsePolicyBuilder) {
+	responsePolicyBuilders = append(responsePolicyBuilders, builder)
+}
+
+func newResponsePolicy(remoteCapabilityInfo commoncap.CapabilityInfo, capabilityMethod string) responsePolicy {
+	for _, builder := range responsePolicyBuilders {
+		policy := builder(remoteCapabilityInfo, capabilityMethod)
+		if policy != nil {
+			return policy
+		}
+	}
+	return nil
+}