Merge pull request #1203 from lightninglabs/block-proof-courier-connect

Block on Proof Courier Service Connection Attempt

Author: guggero

itest/tapd_harness.go

@@ -270,7 +270,8 @@ func newTapdHarness(t *testing.T, ht *harnessTest, cfg tapdConfig,
 		BackoffCfg:         &hashmailBackoffCfg,
 	}
 	finalCfg.UniverseRpcCourier = &proof.UniverseRpcCourierCfg{
-		BackoffCfg: &universeRpcBackoffCfg,
+		BackoffCfg:            &universeRpcBackoffCfg,
+		ServiceRequestTimeout: 50 * time.Millisecond,
 	}
 
 	switch typedProofCourier := (opts.proofCourier).(type) {

itest/test_harness.go

@@ -327,6 +327,8 @@ func setupHarnesses(t *testing.T, ht *harnessTest,
 	// If nothing is specified, we use the universe RPC proof courier by
 	// default.
 	default:
+		t.Logf("Address of universe server as proof courier: %v",
+			universeServer.service.rpcHost())
 		proofCourier = universeServer
 	}
 

proof/courier.go

@@ -654,7 +654,7 @@ func (b *BackoffHandler) Exec(ctx context.Context, proofLocator Locator,
 	if err != nil {
 		return err
 	}
-	log.Infof("Starting proof transfer backoff procedure for proof "+
+	log.Infof("Starting proof transfer backoff procedure "+
 		"(transfer_type=%s, locator_hash=%x)", transferType,
 		locatorHash[:])
 
@@ -710,7 +710,7 @@ func (b *BackoffHandler) Exec(ctx context.Context, proofLocator Locator,
 		)
 		subscriberEvent(waitEvent)
 
-		log.Debugf("Proof delivery failed with error. Backing off. "+
+		log.Debugf("Proof transfer failed with error. Backing off. "+
 			"(transfer_type=%s, locator_hash=%x, backoff=%s, "+
 			"attempt=%d): %v",
 			transferType, locatorHash[:], backoff, i, errExec)
@@ -742,7 +742,7 @@ func (b *BackoffHandler) wait(ctx context.Context, wait time.Duration) error {
 	case <-time.After(wait):
 		return nil
 	case <-ctx.Done():
-		return fmt.Errorf("context canceled")
+		return fmt.Errorf("back off handler context done")
 	}
 }
 
@@ -1154,10 +1154,17 @@ func (h *HashMailCourier) SetSubscribers(
 var _ Courier = (*HashMailCourier)(nil)
 
 // UniverseRpcCourierCfg is the config for the universe RPC proof courier.
+//
+// nolint:lll
 type UniverseRpcCourierCfg struct {
 	// BackoffCfg configures the behaviour of the proof delivery
 	// functionality.
 	BackoffCfg *BackoffCfg
+
+	// ServiceRequestTimeout defines the maximum duration we'll wait for
+	// a courier service to handle our outgoing request during a connection
+	// attempt, or when delivering or retrieving a proof.
+	ServiceRequestTimeout time.Duration `long:"servicerequestimeout" description:"The maximum duration we'll wait for a courier service to handle our outgoing request during a connection attempt, or when delivering or retrieving a proof."`
 }
 
 // UniverseRpcCourier is a universe RPC proof courier service handle. It
@@ -1356,18 +1363,29 @@ func (c *UniverseRpcCourier) DeliverProof(ctx context.Context,
 		deliverFunc := func() error {
 			// Connect to the courier service if a connection hasn't
 			// been established yet.
-			err := c.ensureConnect(ctx)
+			subCtx, subCtxCancel := context.WithTimeout(
+				ctx, c.cfg.ServiceRequestTimeout,
+			)
+			defer subCtxCancel()
+
+			err := c.ensureConnect(subCtx)
 			if err != nil {
 				return fmt.Errorf("unable to connect to "+
 					"courier service during delivery "+
 					"attempt: %w", err)
 			}
 
 			// Submit proof to courier.
-			_, err = c.client.InsertProof(ctx, &unirpc.AssetProof{
+			subCtx, subCtxCancel = context.WithTimeout(
+				ctx, c.cfg.ServiceRequestTimeout,
+			)
+			defer subCtxCancel()
+
+			assetProof := unirpc.AssetProof{
 				Key:       &universeKey,
 				AssetLeaf: &assetLeaf,
-			})
+			}
+			_, err = c.client.InsertProof(subCtx, &assetProof)
 			if err != nil {
 				return fmt.Errorf("error inserting proof "+
 					"into universe courier service: %w",
@@ -1420,15 +1438,25 @@ func (c *UniverseRpcCourier) ReceiveProof(ctx context.Context,
 		receiveFunc := func() error {
 			// Connect to the courier service if a connection hasn't
 			// been established yet.
-			err := c.ensureConnect(ctx)
+			subCtx, subCtxCancel := context.WithTimeout(
+				ctx, c.cfg.ServiceRequestTimeout,
+			)
+			defer subCtxCancel()
+
+			err := c.ensureConnect(subCtx)
 			if err != nil {
 				return fmt.Errorf("unable to connect to "+
-					"courier service during delivery "+
-					"attempt: %w", err)
+					"universe RPC courier service during "+
+					"recieve attempt: %w", err)
 			}
 
 			// Retrieve proof from courier.
-			resp, err := c.client.QueryProof(ctx, &universeKey)
+			subCtx, subCtxCancel = context.WithTimeout(
+				ctx, c.cfg.ServiceRequestTimeout,
+			)
+			defer subCtxCancel()
+
+			resp, err := c.client.QueryProof(subCtx, &universeKey)
 			if err != nil {
 				return fmt.Errorf("error retrieving proof "+
 					"from universe courier service: %w",

sample-tapd.conf

@@ -64,6 +64,11 @@
 ; {s, m, h}.
 ; custodianproofretrievaldelay=5s
 
+; The maximum duration we'll wait for a proof courier service to handle our
+; outgoing request during a connection attempt, or when delivering or retrieving
+; a proof.
+; universerpccourier.servicerequestimeout=5s
+
 ; Network to run on (mainnet, regtest, testnet, simnet, signet)
 ; network=testnet
 

tapcfg/config.go

@@ -98,6 +98,10 @@ const (
 	// use for waiting for a receiver to acknowledge a proof transfer.
 	defaultProofTransferReceiverAckTimeout = time.Hour * 6
 
+	// defaultProofCourierServiceResponseTimeout is the default timeout
+	// we'll use for waiting for a response from the proof courier service.
+	defaultProofCourierServiceResponseTimeout = time.Second * 5
+
 	// defaultUniverseSyncInterval is the default interval that we'll use
 	// to sync Universe state with the federation.
 	defaultUniverseSyncInterval = time.Minute * 10
@@ -435,6 +439,7 @@ func DefaultConfig() Config {
 				InitialBackoff:   defaultProofTransferInitialBackoff,
 				MaxBackoff:       defaultProofTransferMaxBackoff,
 			},
+			ServiceRequestTimeout: defaultProofCourierServiceResponseTimeout,
 		},
 		CustodianProofRetrievalDelay: defaultProofRetrievalDelay,
 		Universe: &UniverseConfig{

tapfreighter/chain_porter.go

@@ -881,6 +881,10 @@ func (p *ChainPorter) transferReceiverProof(pkg *sendPackage) error {
 	// those that do not.
 	reportProofTransfers(notDeliveringOutputs, pendingDeliveryOutputs)
 
+	// incompleteDelivery is set to true if any proof delivery attempts fail
+	// and exceed the maximum backoff limit.
+	incompleteDelivery := false
+
 	deliver := func(ctx context.Context, out TransferOutput) error {
 		scriptKey := out.ScriptKey.PubKey
 		scriptKeyBytes := scriptKey.SerializeCompressed()
@@ -940,6 +944,14 @@ func (p *ChainPorter) transferReceiverProof(pkg *sendPackage) error {
 		// later.
 		var backoffExecErr *proof.BackoffExecError
 		if errors.As(err, &backoffExecErr) {
+			log.Debugf("Exceeded backoff limit for proof delivery "+
+				"(script_key=%x, proof_courier_addr=%s)",
+				scriptKey.SerializeCompressed(),
+				out.ProofCourierAddr)
+
+			// Set the incomplete delivery flag to true so that we
+			// can retry the proof transfer state later.
+			incompleteDelivery = true
 			return nil
 		}
 		if err != nil {
@@ -957,6 +969,10 @@ func (p *ChainPorter) transferReceiverProof(pkg *sendPackage) error {
 				"confirmation: %w", err)
 		}
 
+		log.Infof("Transfer output proof delivery complete "+
+			"(anchor_txid=%v, output_position=%d)",
+			pkg.OutboundPkg.AnchorTx.TxHash(), out.Position)
+
 		return nil
 	}
 
@@ -999,6 +1015,17 @@ func (p *ChainPorter) transferReceiverProof(pkg *sendPackage) error {
 		return firstErr
 	}
 
+	// If the delivery is incomplete, we'll return early so that we can
+	// retry proof transfer later.
+	if incompleteDelivery {
+		log.Debugf("Proof delivery incomplete, will retry executing "+
+			"the proof transfer state (transfer_anchor_tx_hash=%v)",
+			pkg.OutboundPkg.AnchorTx.TxHash())
+
+		// Return here before setting the transfer to complete.
+		return nil
+	}
+
 	// At this point, the transfer is fully finalised and successful:
 	// - The anchoring transaction has been confirmed on-chain.
 	// - The proof(s) have been delivered to the receiver(s).