feat: Balance Manager

alertmanager/alerts.go

@@ -345,12 +345,17 @@ func (al *alerts) getAddresses() ([]address.Address, []address.Address, error) {
 			if strings.Contains(err.Error(), sql.ErrNoRows.Error()) {
 				return nil, nil, xerrors.Errorf("missing layer '%s' ", layer)
 			}
-			return nil, nil, fmt.Errorf("could not read layer '%s': %w", layer, err)
+			return nil, nil, xerrors.Errorf("could not read layer '%s': %w", layer, err)
+		}
+
+		err = config.FixTOML(text, cfg)
+		if err != nil {
+			return nil, nil, err
 		}
 
 		_, err = toml.Decode(text, cfg)
 		if err != nil {
-			return nil, nil, fmt.Errorf("could not read layer, bad toml %s: %w", layer, err)
+			return nil, nil, xerrors.Errorf("could not read layer, bad toml %s: %w", layer, err)
 		}
 
 		for i := range cfg.Addresses {

cmd/curio/config.go

@@ -10,7 +10,6 @@ import (
 	"path"
 	"strings"
 
-	"github.com/BurntSushi/toml"
 	"github.com/fatih/color"
 	"github.com/urfave/cli/v2"
 	"golang.org/x/xerrors"
@@ -292,7 +291,7 @@ var configEditCmd = &cli.Command{
 
 		if cctx.IsSet("source") && source != layer && !cctx.Bool("no-interpret-source") {
 			curioCfg := config.DefaultCurioConfig()
-			if _, err := toml.Decode(sourceConfig, curioCfg); err != nil {
+			if _, err := deps.LoadConfigWithUpgrades(sourceConfig, curioCfg); err != nil {
 				return xerrors.Errorf("parsing source config: %w", err)
 			}
 
@@ -358,12 +357,12 @@ func diff(sourceConf, newConf string) (string, error) {
 	fromSrc := config.DefaultCurioConfig()
 	fromNew := config.DefaultCurioConfig()
 
-	_, err := toml.Decode(sourceConf, fromSrc)
+	_, err := deps.LoadConfigWithUpgrades(sourceConf, fromSrc)
 	if err != nil {
 		return "", xerrors.Errorf("decoding source config: %w", err)
 	}
 
-	_, err = toml.Decode(newConf, fromNew)
+	_, err = deps.LoadConfigWithUpgrades(newConf, fromNew)
 	if err != nil {
 		return "", xerrors.Errorf("decoding new config: %w", err)
 	}

cmd/curio/config_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"fmt"
 	"reflect"
 	"testing"
 	"time"
@@ -555,14 +556,17 @@ func TestConfig(t *testing.T) {
 	addr1 := config.CurioAddresses{
 		PreCommitControl:      []string{},
 		CommitControl:         []string{},
+		DealPublishControl:    []string{},
 		TerminateControl:      []string{"t3qroiebizgkz7pvj26reg5r5mqiftrt5hjdske2jzjmlacqr2qj7ytjncreih2mvujxoypwpfusmwpipvxncq"},
 		DisableOwnerFallback:  false,
 		DisableWorkerFallback: false,
 		MinerAddresses:        []string{"t01000"},
+		BalanceManager:        config.DefaultBalanceManager(),
 	}
 
 	addr2 := config.CurioAddresses{
 		MinerAddresses: []string{"t01001"},
+		BalanceManager: config.DefaultBalanceManager(),
 	}
 
 	_, err := deps.LoadConfigWithUpgrades(baseText, baseCfg)
@@ -604,3 +608,66 @@ func TestCustomConfigDurationJson(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, prop.Type, "string")
 }
+
+func TestTOMLDecoding(t *testing.T) {
+
+	def1 := config.DefaultCurioConfig()
+
+	text := `
+[[Addresses]]
+  MinerAddresses = ["t01002"]
+  [Addresses.BalanceManager]
+	[Addresses.BalanceManager.MK12Collateral]
+	  CollateralLowThreshold = "100 FIL"
+
+[[Addresses]]
+  MinerAddresses = ["t01007"]
+  [Addresses.BalanceManager]
+	[Addresses.BalanceManager.MK12Collateral]
+	  CollateralLowThreshold = "50 fil"
+
+[Alerting]
+  [Alerting.PagerDuty]
+  [Alerting.PrometheusAlertManager]
+  [Alerting.SlackWebhook]
+
+[Apis]
+  ChainApiInfo = ["eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJBbGxvdyI6WyJyZWFkIiwid3JpdGUiLCJzaWduIiwiYWRtaW4iXX0.Er4BD7iisZ6KwdkjbXrxRlgvOnf_KClzo9Q6V7fvUYs:/dns/lotus/tcp/1234/http"]
+  StorageRPCSecret = "E/RAavP4YYHzKqa+eGXE6LtYTrT5YpToCEHugbTXOoI="
+
+[Batching]
+  [Batching.Commit]
+  [Batching.PreCommit]
+  [Batching.Update]
+
+[Fees]
+  [Fees.MaxCommitBatchGasFee]
+  [Fees.MaxPreCommitBatchGasFee]
+  [Fees.MaxUpdateBatchGasFee]
+
+[HTTP]
+  [HTTP.CompressionLevels]
+
+[Ingest]
+
+[Market]
+  [Market.StorageMarketConfig]
+	[Market.StorageMarketConfig.IPNI]
+	[Market.StorageMarketConfig.Indexing]
+	[Market.StorageMarketConfig.MK12]
+
+[Proving]
+
+[Seal]
+
+[Subsystems]
+`
+
+	_, err := deps.LoadConfigWithUpgrades(text, def1)
+	require.NoError(t, err)
+
+	cb, err := config.ConfigUpdate(def1, config.DefaultCurioConfig(), config.Commented(true), config.DefaultKeepUncommented(), config.NoEnv())
+	require.NoError(t, err)
+
+	fmt.Println(string(cb))
+}

cmd/curio/guidedsetup/guidedsetup.go

@@ -627,6 +627,7 @@ func stepNewMinerConfig(d *MigrationData) {
 		DisableOwnerFallback:  false,
 		DisableWorkerFallback: false,
 		MinerAddresses:        []string{d.MinerID.String()},
+		BalanceManager:        config.DefaultBalanceManager(),
 	})
 
 	sk, err := io.ReadAll(io.LimitReader(rand.Reader, 32))

cmd/curio/guidedsetup/shared.go

@@ -158,6 +158,7 @@ func SaveConfigToLayerMigrateSectors(db *harmonydb.DB, minerRepoPath, chainApiIn
 		TerminateControl:      smCfg.Addresses.TerminateControl,
 		DisableOwnerFallback:  smCfg.Addresses.DisableOwnerFallback,
 		DisableWorkerFallback: smCfg.Addresses.DisableWorkerFallback,
+		BalanceManager:        config.DefaultBalanceManager(),
 	}}
 
 	ks, err := lr.KeyStore()

cmd/curio/tasks/tasks.go

@@ -216,6 +216,14 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 		miners = append(miners, address.Address(k))
 	}
 
+	if cfg.Subsystems.EnableBalanceManager {
+		balMgrTask, err := storage_market.NewBalanceManager(full, miners, cfg, sender)
+		if err != nil {
+			return nil, err
+		}
+		activeTasks = append(activeTasks, balMgrTask)
+	}
+
 	{
 		// Market tasks
 		if cfg.Subsystems.EnableDealMarket {

deps/config/load.go

@@ -67,7 +67,21 @@ func FromReader(reader io.Reader, def interface{}, opts ...LoadCfgOpt) (interfac
 		return nil, err
 	}
 	cfg := def
-	md, err := toml.NewDecoder(reader).Decode(cfg)
+	var buf bytes.Buffer
+	_, err = io.Copy(&buf, reader)
+	if err != nil {
+		return toml.MetaData{}, err
+	}
+
+	if ccfg, ok := cfg.(*CurioConfig); ok {
+		err = FixTOML(buf.String(), ccfg)
+		if err != nil {
+			return nil, err
+		}
+		cfg = ccfg
+	}
+
+	md, err := toml.Decode(buf.String(), cfg)
 	if err != nil {
 		return nil, err
 	}
@@ -499,3 +513,36 @@ func findDocSect(root, section, name string) *DocField {
 
 	return nil
 }
+
+func FixTOML(newText string, cfg *CurioConfig) error {
+	// This is a workaround to set the length of [[Addresses]] correctly before we do toml.Decode.
+	// The reason this is required is that toml libraries create nil pointer to uninitialized structs.
+	// This in turn causes failure to decode types like types.FIL which are struct with unexported pointer inside
+	type AddressLengthDetector struct {
+		Addresses []struct{} `toml:"Addresses"`
+	}
+
+	var lengthDetector AddressLengthDetector
+	_, err := toml.Decode(newText, &lengthDetector)
+	if err != nil {
+		return xerrors.Errorf("Error decoding TOML for length detection: %w", err)
+	}
+
+	l := len(lengthDetector.Addresses)
+	il := len(cfg.Addresses)
+
+	for l > il {
+		cfg.Addresses = append(cfg.Addresses, CurioAddresses{
+			PreCommitControl:      []string{},
+			CommitControl:         []string{},
+			DealPublishControl:    []string{},
+			TerminateControl:      []string{},
+			DisableOwnerFallback:  false,
+			DisableWorkerFallback: false,
+			MinerAddresses:        []string{},
+			BalanceManager:        DefaultBalanceManager(),
+		})
+		il++
+	}
+	return nil
+}

deps/config/types.go

@@ -42,6 +42,7 @@ func DefaultCurioConfig() *CurioConfig {
 			DealPublishControl: []string{},
 			TerminateControl:   []string{},
 			MinerAddresses:     []string{},
+			BalanceManager:     DefaultBalanceManager(),
 		}},
 		Proving: CurioProvingConfig{
 			ParallelCheckLimit:    32,
@@ -131,6 +132,16 @@ func DefaultCurioConfig() *CurioConfig {
 	}
 }
 
+func DefaultBalanceManager() BalanceManagerConfig {
+	return BalanceManagerConfig{
+		MK12Collateral: MK12CollateralConfig{
+			DealCollateralWallet:    "",
+			CollateralLowThreshold:  types.MustParseFIL("5 FIL"),
+			CollateralHighThreshold: types.MustParseFIL("20 FIL"),
+		},
+	}
+}
+
 // CurioConfig defines configuration for the Curio node
 type CurioConfig struct {
 
@@ -367,6 +378,9 @@ type CurioSubsystemsConfig struct {
 	// also be bounded by resources available on the machine. (Default: 8)
 	IndexingMaxTasks int
 
+	// EnableBalanceManager enables the task to automatically manage the market balance of the miner's market actor (Default: false)
+	EnableBalanceManager bool
+
 	// BindSDRTreeToNode forces the TreeD and TreeRC tasks to be executed on the same node where SDR task was executed
 	// for the sector. Please ensure that TreeD and TreeRC task are enabled and relevant resources are available before
 	// enabling this option. (Default: false)
@@ -421,6 +435,10 @@ type CurioAddresses struct {
 
 	// MinerAddresses are the addresses of the miner actors
 	MinerAddresses []string
+
+	// BalanceManagerConfig specifies the configuration parameters for managing wallet balances and actor-related funds,
+	// including collateral and other operational resources.
+	BalanceManager BalanceManagerConfig
 }
 
 type CurioProvingConfig struct {
@@ -808,3 +826,23 @@ type CompressionConfig struct {
 	BrotliLevel  int
 	DeflateLevel int
 }
+
+type BalanceManagerConfig struct {
+	// MK12Collateral defines the configuration for managing collateral and related balance thresholds in the miner's market.
+	MK12Collateral MK12CollateralConfig
+}
+
+type MK12CollateralConfig struct {
+	// DealCollateralWallet is the wallet used to add balance to Miner's market balance. This balance is
+	// utilized for deal collateral in market (f05) deals.
+	DealCollateralWallet string
+
+	// CollateralLowThreshold is the balance below which more balance will be added to miner's market balance
+	// Accepts a decimal string (e.g., "123.45" or "123 fil") with optional "fil" or "attofil" suffix. (Default: "5 FIL")
+	CollateralLowThreshold types.FIL
+
+	// CollateralHighThreshold is the target balance to which the miner's market balance will be topped up
+	// when it drops below CollateralLowThreshold.
+	// Accepts a decimal string (e.g., "123.45" or "123 fil") with optional "fil" or "attofil" suffix. (Default: "20 FIL")
+	CollateralHighThreshold types.FIL
+}