feat: celestia fee should be calculated dynamically (#417)

config/toml.go

@@ -85,7 +85,7 @@ da_config = "{{ .DAConfig }}"
 block_batch_max_size_bytes = {{ .BlockManagerConfig.BlockBatchMaxSizeBytes }}
 
 #celestia config example:
-# da_config = "{\"base_url\": \"http://127.0.0.1:26659\", \"timeout\": 60000000000, \"fee\":20000, \"gas_limit\": 20000000, \"namespace_id\":\"000000000000ffff\"}"
+# da_config = "{\"base_url\": \"http://127.0.0.1:26659\", \"timeout\": 60000000000, \"gas_prices\":0.1, \"gas_limit\": 20000000, \"namespace_id\":\"000000000000ffff\"}"
 # Avail config example:
 # da_config = "{\"seed\": \"MNEMONIC\", \"api_url\": \"wss://kate.avail.tools/ws\", \"app_id\": 0, \"tip\":10}"
 

da/celestia/celestia.go

@@ -13,7 +13,7 @@ import (
 	rpcclient "github.com/tendermint/tendermint/rpc/client"
 	httprpcclient "github.com/tendermint/tendermint/rpc/client/http"
 
-	"github.com/celestiaorg/go-cnc"
+	cnc "github.com/celestiaorg/go-cnc"
 	"github.com/dymensionxyz/dymint/da"
 	"github.com/dymensionxyz/dymint/log"
 	"github.com/dymensionxyz/dymint/store"
@@ -95,6 +95,14 @@ func (c *DataAvailabilityLayerClient) Init(config []byte, pubsubServer *pubsub.S
 		return err
 	}
 
+	if c.config.GasPrices != 0 && c.config.Fee != 0 {
+		return errors.New("can't set both gas prices and fee")
+	}
+
+	if c.config.Fee == 0 && c.config.GasPrices == 0 {
+		return errors.New("fee or gas prices must be set")
+	}
+
 	c.pubsubServer = pubsubServer
 	// Set defaults
 	c.txPollingRetryDelay = defaultTxPollingRetryDelay
@@ -159,62 +167,61 @@ func (c *DataAvailabilityLayerClient) SubmitBatch(batch *types.Batch) da.ResultS
 			c.logger.Debug("Context cancelled")
 			return da.ResultSubmitBatch{}
 		default:
-			txResponse, err := c.client.SubmitPFB(c.ctx, c.config.NamespaceID, blob, c.config.Fee, c.config.GasLimit)
-			if txResponse != nil {
-				if txResponse.Code != 0 {
-					c.logger.Debug("Failed to submit DA batch. Emitting health event and trying again", "txResponse", txResponse, "error", err)
-					// Publish an health event. Only if we failed to emit the event we return an error.
-					res, err := da.SubmitBatchHealthEventHelper(c.pubsubServer, c.ctx, false, errors.New(txResponse.RawLog))
-					if err != nil {
-						return res
-					}
-				} else if err != nil {
-					// Here we assume that if txResponse is not nil and also error is not nil it means that the transaction
-					// was submitted (not necessarily accepted) and we still didn't get a clear status regarding it (e.g timeout).
-					// hence trying to poll for it.
-					c.logger.Debug("Failed to receive DA batch inclusion result. Waiting for inclusion", "txResponse", txResponse.RawLog, "error", err)
-					inclusionHeight, err := c.waitForTXInclusion(txResponse.TxHash)
-					if err == nil {
-						res, err := da.SubmitBatchHealthEventHelper(c.pubsubServer, c.ctx, true, nil)
-						if err != nil {
-							return res
-						} else {
-							return da.ResultSubmitBatch{
-								BaseResult: da.BaseResult{
-									Code:     da.StatusSuccess,
-									Message:  "tx hash: " + txResponse.TxHash,
-									DAHeight: inclusionHeight,
-								},
-							}
-						}
-					} else {
-						c.logger.Debug("Failed to receive DA batch inclusion result. Emitting health event and trying again", "error", err)
-						res, err := da.SubmitBatchHealthEventHelper(c.pubsubServer, c.ctx, false, err)
-						if err != nil {
-							return res
-						}
-					}
-
-				} else {
-					c.logger.Debug("Successfully submitted DA batch", "txResponse", txResponse.RawLog)
-					res, err := da.SubmitBatchHealthEventHelper(c.pubsubServer, c.ctx, true, nil)
-					if err != nil {
-						return res
-					}
-					return da.ResultSubmitBatch{
-						BaseResult: da.BaseResult{
-							Code:     da.StatusSuccess,
-							Message:  "tx hash: " + txResponse.TxHash,
-							DAHeight: uint64(txResponse.Height),
-						},
-					}
+			estimatedGas := EstimateGas(len(blob))
+			gasWanted := uint64(float64(estimatedGas) * gasAdjustment)
+			fees := c.calculateFees(gasWanted)
+
+			//SubmitPFB sets an error if the txResponse has error, so we check check the txResponse for error
+			txResponse, err := c.client.SubmitPFB(c.ctx, c.config.NamespaceID, blob, int64(fees), gasWanted)
+			if txResponse == nil {
+				c.logger.Error("Failed to submit DA batch. Emitting health event and trying again", "error", err)
+				res, err := da.SubmitBatchHealthEventHelper(c.pubsubServer, c.ctx, false, err)
+				if err != nil {
+					return res
 				}
-			} else {
-				res, err := da.SubmitBatchHealthEventHelper(c.pubsubServer, c.ctx, false, errors.New("DA txResponse is nil"))
+				time.Sleep(c.submitRetryDelay)
+				continue
+			}
+
+			if txResponse.Code != 0 {
+				c.logger.Error("Failed to submit DA batch. Emitting health event and trying again", "txResponse", txResponse.RawLog, "code", txResponse.Code)
+				res, err := da.SubmitBatchHealthEventHelper(c.pubsubServer, c.ctx, false, errors.New(txResponse.RawLog))
 				if err != nil {
 					return res
 				}
 				time.Sleep(c.submitRetryDelay)
+				continue
+			}
+
+			// Here we assume that if txResponse is not nil and also error is not nil it means that the transaction
+			// was submitted (not necessarily accepted) and we still didn't get a clear status regarding it (e.g timeout).
+			// hence trying to poll for it.
+			daHeight := uint64(txResponse.Height)
+			if daHeight == 0 {
+				c.logger.Debug("Failed to receive DA batch inclusion result. Waiting for inclusion", "txHash", txResponse.TxHash)
+				daHeight, err = c.waitForTXInclusion(txResponse.TxHash)
+				if err != nil {
+					c.logger.Error("Failed to receive DA batch inclusion result. Emitting health event and trying again", "error", err)
+					res, err := da.SubmitBatchHealthEventHelper(c.pubsubServer, c.ctx, false, err)
+					if err != nil {
+						return res
+					}
+					time.Sleep(c.submitRetryDelay)
+					continue
+				}
+			}
+
+			c.logger.Info("Successfully submitted DA batch", "txHash", txResponse.TxHash, "daHeight", txResponse.Height, "gasWanted", txResponse.GasWanted, "gasUsed", txResponse.GasUsed)
+			res, err := da.SubmitBatchHealthEventHelper(c.pubsubServer, c.ctx, true, nil)
+			if err != nil {
+				return res
+			}
+			return da.ResultSubmitBatch{
+				BaseResult: da.BaseResult{
+					Code:     da.StatusSuccess,
+					Message:  "tx hash: " + txResponse.TxHash,
+					DAHeight: daHeight,
+				},
 			}
 		}
 	}

da/celestia/celestia_test.go

@@ -31,7 +31,7 @@ const (
 func TestSubmitBatch(t *testing.T) {
 	assert := assert.New(t)
 	require := require.New(t)
-	configBytes, err := json.Marshal(celestia.Config{})
+	configBytes, err := json.Marshal(celestia.CelestiaDefaultConfig)
 	require.NoError(err)
 	batch := &types.Batch{
 		StartHeight: 0,

da/celestia/config.go

@@ -4,14 +4,16 @@ import (
 	"encoding/hex"
 	"time"
 
-	"github.com/celestiaorg/go-cnc"
+	cnc "github.com/celestiaorg/go-cnc"
 )
 
 const (
 	defaultTxPollingRetryDelay = 20 * time.Second
 	defaultSubmitRetryDelay    = 10 * time.Second
 	defaultTxPollingAttempts   = 5
 	namespaceVersion           = 0
+	defaultGasPrices           = 0.1
+	gasAdjustment              = 1.3
 )
 
 // Config stores Celestia DALC configuration parameters.
@@ -20,6 +22,7 @@ type Config struct {
 	AppNodeURL     string        `json:"app_node_url"`
 	Timeout        time.Duration `json:"timeout"`
 	Fee            int64         `json:"fee"`
+	GasPrices      float64       `json:"gas_prices"`
 	GasLimit       uint64        `json:"gas_limit"`
 	NamespaceIDStr string        `json:"namespace_id"`
 	NamespaceID    cnc.Namespace `json:"-"`
@@ -29,8 +32,9 @@ var CelestiaDefaultConfig = Config{
 	BaseURL:        "http://127.0.0.1:26659",
 	AppNodeURL:     "",
 	Timeout:        30 * time.Second,
-	Fee:            20000,
+	Fee:            0,
 	GasLimit:       20000000,
+	GasPrices:      defaultGasPrices,
 	NamespaceIDStr: "000000000000ffff",
 	NamespaceID:    cnc.Namespace{Version: namespaceVersion, ID: []byte{0, 0, 0, 0, 0, 0, 255, 255}},
 }

da/celestia/fees.go

@@ -0,0 +1,22 @@
+package celestia
+
+const (
+	perByteGasTolerance   = 2
+	pfbGasFixedCost       = 80000
+	defaultGasPerBlobByte = 8
+)
+
+func (c *DataAvailabilityLayerClient) calculateFees(gas uint64) int64 {
+	fees := c.config.Fee
+	if fees == 0 {
+		fees = int64(c.config.GasPrices * float64(gas))
+	}
+
+	return fees
+}
+
+// EstimateGas estimates the gas required to pay for a set of blobs in a PFB.
+func EstimateGas(blobSizes int) uint64 {
+	variableGasAmount := (defaultGasPerBlobByte + perByteGasTolerance) * blobSizes
+	return uint64(variableGasAmount + pfbGasFixedCost)
+}

da/da_test.go

@@ -99,6 +99,7 @@ func doTestDALC(t *testing.T, dalc da.DataAvailabilityLayerClient) {
 			BaseURL:     "http://localhost:26658",
 			Timeout:     30 * time.Second,
 			GasLimit:    3000000,
+			Fee:         200000000,
 			NamespaceID: cnc.Namespace{Version: 0, ID: []byte{0, 0, 0, 0, 0, 0, 255, 255}},
 		}
 		conf, _ = json.Marshal(config)
@@ -219,6 +220,7 @@ func doTestRetrieve(t *testing.T, dalc da.DataAvailabilityLayerClient) {
 			BaseURL:     "http://localhost:26658",
 			Timeout:     30 * time.Second,
 			GasLimit:    3000000,
+			Fee:         2000000,
 			NamespaceID: cnc.Namespace{Version: 0, ID: []byte{0, 0, 0, 0, 0, 0, 255, 255}},
 		}
 		conf, _ = json.Marshal(config)