cre aptos: stabilize write tx-hash resolution and enforce failed-write hash verification in suite

Author: cawthorne

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

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -17,6 +18,7 @@ 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"
 
@@ -46,9 +48,18 @@ 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
+
 	respSent bool
 	mux      sync.Mutex
 	wg       *sync.WaitGroup
@@ -183,17 +194,28 @@ 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),
 		responseCh:                 make(chan clientResponse, 1),
 		wg:                         &wg,
 		lggr:                       lggr,
@@ -263,6 +285,14 @@ 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
+		}
 		c.sendResponse(clientResponse{Err: err})
 	}
 }
@@ -330,14 +360,27 @@ func (c *ClientRequest) OnMessage(_ context.Context, msg *types.MessageBody) err
 			lggr.Warnw("received multiple unique responses for the same request", "count for responseID", len(c.responseIDCount))
 		}
 
-		if c.responseIDCount[responseID] == c.requiredIdenticalResponses {
+		if c.responseIDCount[responseID] == c.requiredIdenticalResponses && !c.shouldDeferIdenticalResponse(msg.Payload) {
 			payload, err := c.encodePayloadWithMetadata(msg, commoncap.ResponseMetadata{Metering: nodeReports})
 			if err != nil {
 				return fmt.Errorf("failed to encode payload with metadata: %w", err)
 			}
 
 			c.sendResponse(clientResponse{Result: payload})
 		}
+
+		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
+			}
+		}
 	} else {
 		c.lggr.Debugw("received error from peer", "error", msg.Error, "errorMsg", msg.ErrorMsg, "peer", sender)
 		c.errorCount[msg.ErrorMsg]++
@@ -347,6 +390,13 @@ 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 {
+				return err
+			}
+			return nil
+		}
+
 		if c.errorCount[msg.ErrorMsg] == c.requiredIdenticalResponses {
 			c.sendResponse(clientResponse{Err: fmt.Errorf("%s : %s", msg.Error, msg.ErrorMsg)})
 		} else if c.totalErrorCount == c.remoteNodeCount-c.requiredIdenticalResponses+1 {
@@ -396,3 +446,196 @@ 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 {
+		return false
+	}
+
+	return reply.GetTxStatus() == aptoscap.TxStatus_TX_STATUS_FAILED
+}
+
+func (c *ClientRequest) allResponsesReceived() bool {
+	if len(c.responseReceived) == 0 {
+		return false
+	}
+
+	for _, received := range c.responseReceived {
+		if !received {
+			return false
+		}
+	}
+
+	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() {
+		return nil
+	}
+
+	payload, ok, buildErr := c.buildDeterministicAptosFailedResponse(true)
+	if buildErr != nil {
+		return fmt.Errorf("failed to build deterministic Aptos failed response: %w", buildErr)
+	}
+	if ok {
+		c.sendResponse(clientResponse{Result: payload})
+		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,
+		),
+	})
+	return nil
+}