tapchannelmsg: send+recv chunks of the input proofs

In this commit, we start to send+recv chunks of the input proofs. This
ensures that if a suffix proof is larger than the `lnwire` message size,
then we'll be able to still send+recv it.

Author: Roasbeef

tapchannel/aux_funding_controller.go

@@ -74,6 +74,10 @@ const (
 	// maxNumHTLCsPerParty is the maximum number of HTLCs that can be added
 	// by a single party to a channel.
 	maxNumHTLCsPerParty = maxNumHTLCs / 2
+
+	// proofChunk size is the chunk size of proofs, in the case that a proof
+	// is too large to be sent in a single message.
+	proofChunkSize = 60_000
 )
 
 // ErrorReporter is used to report an error back to the caller and/or peer that
@@ -420,7 +424,9 @@ type pendingAssetFunding struct {
 	fundingAckChan chan bool
 
 	fundingFinalizedSignal chan struct{}
-	finalizedCloseOnce     sync.Once
+
+	finalizedCloseOnce sync.Once
+	inputProofChunks   map[chainhash.Hash][]cmsg.ProofChunk
 }
 
 // addInputProof adds a new proof to the set of proofs that'll be used to fund
@@ -461,6 +467,34 @@ func (p *pendingAssetFunding) addToFundingCommitment(a *asset.Asset) error {
 	return p.fundingAssetCommitment.Merge(newCommitment)
 }
 
+// addInputProofChunk adds a new proof chunk to the set of proof chunks that'll
+// be processed. If this is the last chunk for this proof, then true is
+// returned.
+func (p *pendingAssetFunding) addInputProofChunk(chunk cmsg.ProofChunk,
+) (*proof.Proof, error) {
+
+	// Collect this proof chunk with the rest of the proofs.
+	chunkID := chunk.ChunkSumID.Val
+
+	proofChunks := p.inputProofChunks[chunkID]
+	proofChunks = append(proofChunks, chunk)
+	p.inputProofChunks[chunkID] = proofChunks
+
+	// If this isn't the last chunk, then we can just return nil and exit
+	if !chunk.Last.Val {
+		return nil, nil
+	}
+
+	// Otherwise, this is the last chunk, so we'll extract all the chunks
+	// and assemble the final proof.
+	finalProof, err := cmsg.AssembleProofChunks(proofChunks)
+	if err != nil {
+		return nil, fmt.Errorf("unable to assemble proof chunks: %w", err)
+	}
+
+	return finalProof, nil
+}
+
 // newCommitBlobAndLeaves creates a new commitment blob that'll be stored in
 // the channel state for the specified party.
 func newCommitBlobAndLeaves(pendingFunding *pendingAssetFunding,
@@ -699,6 +733,7 @@ func (f *fundingFlowIndex) fromMsg(chainParams *address.ChainParams,
 			amt:                    assetProof.Amt().UnwrapOr(0),
 			fundingAckChan:         make(chan bool, 1),
 			fundingFinalizedSignal: make(chan struct{}),
+			inputProofChunks:       make(map[chainhash.Hash][]cmsg.ProofChunk),
 		}
 		(*f)[pid] = assetFunding
 	}
@@ -827,15 +862,33 @@ func (f *FundingController) sendInputOwnershipProofs(peerPub btcec.PublicKey,
 		log.Tracef("Sending input ownership proof to remote party: %x",
 			proofBytes)
 
-		inputProof := cmsg.NewTxAssetInputProof(
-			fundingState.pid, *fundingState.inputProofs[i],
-		)
+		inputProof := fundingState.inputProofs[i]
+		inputAsset := inputProof.Asset
 
-		// Finally, we'll send the proof to the remote peer.
-		err := f.cfg.PeerMessenger.SendMessage(ctx, peerPub, inputProof)
+		// For each proof, we'll chunk them up optimistically to make
+		// sure we'll never exceed the upper message limit.
+		proofChunks, err := cmsg.CreateProofChunks(
+			*inputProof, proofChunkSize,
+		)
 		if err != nil {
-			return fmt.Errorf("unable to send proof to peer: %w",
-				err)
+			return fmt.Errorf("unable to create proof "+
+				"chunks: %w", err)
+		}
+
+		for _, proofChunk := range proofChunks {
+			inputProof := cmsg.NewTxAssetInputProof(
+				fundingState.pid, inputAsset.ID(),
+				inputAsset.Amount, proofChunk,
+			)
+
+			// Finally, we'll send the proof to the remote peer.
+			err := f.cfg.PeerMessenger.SendMessage(
+				ctx, peerPub, inputProof,
+			)
+			if err != nil {
+				return fmt.Errorf("unable to send "+
+					"proof to peer: %w", err)
+			}
 		}
 	}
 
@@ -1295,9 +1348,27 @@ func (f *FundingController) processFundingMsg(ctx context.Context,
 	// This is input proof, so we'll verify the challenge witness, then
 	// store the proof.
 	case *cmsg.TxAssetInputProof:
+		// By default, we'll get chunks of the proof sent to us. So
+		// we'll add this set to the chunks, then proceed but only if we
+		// have all the chunks.
+		finalProof, err := assetFunding.addInputProofChunk(
+			assetProof.ProofChunk.Val,
+		)
+		if err != nil {
+			return tempPID, fmt.Errorf("unable to add input proof "+
+				"chunk: %w", err)
+		}
+
+		// If there's no final proof yet, we can just return>
+		if finalProof == nil {
+			return tempPID, nil
+		}
+
+		// Otherwise, we have all the proofs we need.
+		//
 		// Before we proceed, we'll make sure that we already know of
 		// the genesis proof for the incoming asset.
-		_, err := f.cfg.AssetSyncer.QueryAssetInfo(
+		_, err = f.cfg.AssetSyncer.QueryAssetInfo(
 			ctx, assetProof.AssetID.Val,
 		)
 		if err != nil {
@@ -1306,10 +1377,10 @@ func (f *FundingController) processFundingMsg(ctx context.Context,
 				assetProof.AssetID.Val, err)
 		}
 
-		p := assetProof.Proof.Val
+		p := finalProof
 		log.Infof("Validating input proof, prev_out=%v", p.OutPoint())
 
-		l, err := f.cfg.ChainBridge.GenProofChainLookup(&p)
+		l, err := f.cfg.ChainBridge.GenProofChainLookup(p)
 		if err != nil {
 			return tempPID, fmt.Errorf("unable to create proof "+
 				"lookup: %w", err)
@@ -1331,7 +1402,7 @@ func (f *FundingController) processFundingMsg(ctx context.Context,
 		// Now that we know the proof is valid, we'll add it to the
 		// funding state.
 		assetFunding.addInputProof(
-			&assetProof.Proof.Val,
+			finalProof,
 		)
 
 	// This is an output proof, so now we should be able to verify the