Config for deal publishing control addresses

api/api_storage.go

@@ -279,15 +279,17 @@ type AddrUse int
 const (
 	PreCommitAddr AddrUse = iota
 	CommitAddr
+	DealPublishAddr
 	PoStAddr
 
 	TerminateSectorsAddr
 )
 
 type AddressConfig struct {
-	PreCommitControl []address.Address
-	CommitControl    []address.Address
-	TerminateControl []address.Address
+	PreCommitControl   []address.Address
+	CommitControl      []address.Address
+	TerminateControl   []address.Address
+	DealPublishControl []address.Address
 
 	DisableOwnerFallback  bool
 	DisableWorkerFallback bool

chain/actors/policy/policy.go

@@ -196,6 +196,30 @@ func GetMaxProveCommitDuration(ver actors.Version, t abi.RegisteredSealProof) ab
 	}
 }
 
+func SetProviderCollateralSupplyTarget(num, denom big.Int) {
+
+	market2.ProviderCollateralSupplyTarget = builtin2.BigFrac{
+		Numerator:   num,
+		Denominator: denom,
+	}
+
+	market3.ProviderCollateralSupplyTarget = builtin3.BigFrac{
+		Numerator:   num,
+		Denominator: denom,
+	}
+
+	market4.ProviderCollateralSupplyTarget = builtin4.BigFrac{
+		Numerator:   num,
+		Denominator: denom,
+	}
+
+	market5.ProviderCollateralSupplyTarget = builtin5.BigFrac{
+		Numerator:   num,
+		Denominator: denom,
+	}
+
+}
+
 func DealProviderCollateralBounds(
 	size abi.PaddedPieceSize, verified bool,
 	rawBytePower, qaPower, baselinePower abi.StoragePower,

chain/actors/policy/policy.go.template

@@ -132,6 +132,17 @@ func GetMaxProveCommitDuration(ver actors.Version, t abi.RegisteredSealProof) ab
 	}
 }
 
+func SetProviderCollateralSupplyTarget(num, denom big.Int) {
+{{range .versions}}
+	{{if (ge . 2)}}
+	market{{.}}.ProviderCollateralSupplyTarget = builtin{{.}}.BigFrac{
+		Numerator:   num,
+		Denominator: denom,
+	}
+	{{end}}
+{{end}}
+}
+
 func DealProviderCollateralBounds(
 	size abi.PaddedPieceSize, verified bool,
 	rawBytePower, qaPower, baselinePower abi.StoragePower,

cmd/lotus-storage-miner/actor.go

@@ -435,6 +435,7 @@ var actorControlList = &cli.Command{
 		commit := map[address.Address]struct{}{}
 		precommit := map[address.Address]struct{}{}
 		terminate := map[address.Address]struct{}{}
+		dealPublish := map[address.Address]struct{}{}
 		post := map[address.Address]struct{}{}
 
 		for _, ca := range mi.ControlAddresses {
@@ -471,6 +472,16 @@ var actorControlList = &cli.Command{
 			terminate[ca] = struct{}{}
 		}
 
+		for _, ca := range ac.DealPublishControl {
+			ca, err := api.StateLookupID(ctx, ca, types.EmptyTSK)
+			if err != nil {
+				return err
+			}
+
+			delete(post, ca)
+			dealPublish[ca] = struct{}{}
+		}
+
 		printKey := func(name string, a address.Address) {
 			b, err := api.WalletBalance(ctx, a)
 			if err != nil {
@@ -515,6 +526,9 @@ var actorControlList = &cli.Command{
 			if _, ok := terminate[a]; ok {
 				uses = append(uses, color.YellowString("terminate"))
 			}
+			if _, ok := dealPublish[a]; ok {
+				uses = append(uses, color.MagentaString("deals"))
+			}
 
 			tw.Write(map[string]interface{}{
 				"name":    name,

documentation/en/api-v0-methods-miner.md

@@ -227,6 +227,7 @@ Response:
   "PreCommitControl": null,
   "CommitControl": null,
   "TerminateControl": null,
+  "DealPublishControl": null,
   "DisableOwnerFallback": true,
   "DisableWorkerFallback": true
 }

itests/ccupgrade_test.go

@@ -73,9 +73,13 @@ func runTestCCUpgrade(t *testing.T, upgradeHeight abi.ChainEpoch) {
 
 	{
 		exp, err := client.StateSectorExpiration(ctx, maddr, CC, types.EmptyTSK)
-		require.NoError(t, err)
-		require.NotNil(t, exp)
-		require.Greater(t, 50000, int(exp.OnTime))
+		if err != nil {
+			require.Contains(t, err.Error(), "failed to find sector 3") // already cleaned up
+		} else {
+			require.NoError(t, err)
+			require.NotNil(t, exp)
+			require.Greater(t, 50000, int(exp.OnTime))
+		}
 	}
 	{
 		exp, err := client.StateSectorExpiration(ctx, maddr, Upgraded, types.EmptyTSK)

itests/deals_publish_test.go

@@ -11,9 +11,13 @@ import (
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/market"
 	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/chain/wallet"
 	"github.com/filecoin-project/lotus/itests/kit"
 	"github.com/filecoin-project/lotus/markets/storageadapter"
 	"github.com/filecoin-project/lotus/node"
+	"github.com/filecoin-project/lotus/node/config"
+	"github.com/filecoin-project/lotus/node/modules"
+	"github.com/filecoin-project/lotus/storage"
 	market2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/market"
 	"github.com/stretchr/testify/require"
 )
@@ -28,16 +32,34 @@ func TestPublishDealsBatching(t *testing.T) {
 
 	kit.QuietMiningLogs()
 
-	opts := node.Override(new(*storageadapter.DealPublisher),
-		storageadapter.NewDealPublisher(nil, storageadapter.PublishMsgConfig{
-			Period:         publishPeriod,
-			MaxDealsPerMsg: maxDealsPerMsg,
-		}),
+	publisherKey, err := wallet.GenerateKey(types.KTSecp256k1)
+	require.NoError(t, err)
+
+	opts := node.Options(
+		node.Override(new(*storageadapter.DealPublisher),
+			storageadapter.NewDealPublisher(nil, storageadapter.PublishMsgConfig{
+				Period:         publishPeriod,
+				MaxDealsPerMsg: maxDealsPerMsg,
+			}),
+		),
+		node.Override(new(*storage.AddressSelector), modules.AddressSelector(&config.MinerAddressConfig{
+			DealPublishControl: []string{
+				publisherKey.Address.String(),
+			},
+			DisableOwnerFallback:  true,
+			DisableWorkerFallback: true,
+		})),
+		kit.LatestActorsAt(-1),
 	)
 
-	client, miner, ens := kit.EnsembleMinimal(t, kit.MockProofs(), kit.ConstructorOpts(opts))
+	client, miner, ens := kit.EnsembleMinimal(t, kit.Account(publisherKey, types.FromFil(10)), kit.MockProofs(), kit.ConstructorOpts(opts))
 	ens.InterconnectAll().BeginMining(10 * time.Millisecond)
 
+	_, err = client.WalletImport(ctx, &publisherKey.KeyInfo)
+	require.NoError(t, err)
+
+	miner.SetControlAddresses(publisherKey.Address)
+
 	dh := kit.NewDealHarness(t, client, miner)
 
 	// Starts a deal and waits until it's published
@@ -92,6 +114,7 @@ func TestPublishDealsBatching(t *testing.T) {
 			err = pubDealsParams.UnmarshalCBOR(bytes.NewReader(msg.Params))
 			require.NoError(t, err)
 			require.Len(t, pubDealsParams.Deals, int(maxDealsPerMsg))
+			require.Equal(t, publisherKey.Address.String(), msg.From.String())
 		}
 	}
 	require.Equal(t, 1, count)

itests/kit/control.go

@@ -0,0 +1,42 @@
+package kit
+
+import (
+	"context"
+
+	"github.com/stretchr/testify/require"
+
+	addr "github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-state-types/big"
+	miner2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/miner"
+
+	"github.com/filecoin-project/lotus/chain/actors"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/chain/types"
+)
+
+func (tm *TestMiner) SetControlAddresses(addrs ...addr.Address) {
+	ctx := context.TODO()
+
+	mi, err := tm.FullNode.StateMinerInfo(ctx, tm.ActorAddr, types.EmptyTSK)
+	require.NoError(tm.t, err)
+
+	cwp := &miner2.ChangeWorkerAddressParams{
+		NewWorker:       mi.Worker,
+		NewControlAddrs: addrs,
+	}
+
+	sp, err := actors.SerializeParams(cwp)
+	require.NoError(tm.t, err)
+
+	smsg, err := tm.FullNode.MpoolPushMessage(ctx, &types.Message{
+		From:   mi.Owner,
+		To:     tm.ActorAddr,
+		Method: miner.Methods.ChangeWorkerAddress,
+
+		Value:  big.Zero(),
+		Params: sp,
+	}, nil)
+	require.NoError(tm.t, err)
+
+	tm.FullNode.WaitMsg(ctx, smsg.Cid())
+}

itests/kit/funds.go

@@ -4,9 +4,11 @@ import (
 	"context"
 	"testing"
 
-	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/stretchr/testify/require"
 
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/ipfs/go-cid"
+
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -27,8 +29,12 @@ func SendFunds(ctx context.Context, t *testing.T, sender *TestFullNode, recipien
 	sm, err := sender.MpoolPushMessage(ctx, msg, nil)
 	require.NoError(t, err)
 
-	res, err := sender.StateWaitMsg(ctx, sm.Cid(), 3, api.LookbackNoLimit, true)
-	require.NoError(t, err)
+	sender.WaitMsg(ctx, sm.Cid())
+}
+
+func (f *TestFullNode) WaitMsg(ctx context.Context, msg cid.Cid) {
+	res, err := f.StateWaitMsg(ctx, msg, 3, api.LookbackNoLimit, true)
+	require.NoError(f.t, err)
 
-	require.EqualValues(t, 0, res.Receipt.ExitCode, "did not successfully send funds")
+	require.EqualValues(f.t, 0, res.Receipt.ExitCode, "message did not successfully execute")
 }

itests/kit/init.go

@@ -5,6 +5,7 @@ import (
 	"os"
 
 	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/big"
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/actors/policy"
 	logging "github.com/ipfs/go-log/v2"
@@ -13,6 +14,8 @@ import (
 func init() {
 	_ = logging.SetLogLevel("*", "INFO")
 
+	policy.SetProviderCollateralSupplyTarget(big.Zero(), big.NewInt(1))
+
 	policy.SetConsensusMinerMinPower(abi.NewStoragePower(2048))
 	policy.SetSupportedProofTypes(abi.RegisteredSealProof_StackedDrg2KiBV1)
 	policy.SetMinVerifiedDealSize(abi.NewStoragePower(256))

markets/storageadapter/client.go

@@ -160,8 +160,16 @@ func (c *ClientNodeAdapter) ValidatePublishedDeal(ctx context.Context, deal stor
 		return 0, xerrors.Errorf("failed to resolve from msg ID addr: %w", err)
 	}
 
-	if fromid != mi.Worker {
-		return 0, xerrors.Errorf("deal wasn't published by storage provider: from=%s, provider=%s", pubmsg.From, deal.Proposal.Provider)
+	var pubOk bool
+	pubAddrs := append([]address.Address{mi.Worker, mi.Owner}, mi.ControlAddresses...)
+	for _, a := range pubAddrs {
+		if fromid == a {
+			pubOk = true
+			break
+		}
+	}
+	if !pubOk {
+		return 0, xerrors.Errorf("deal wasn't published by storage provider: from=%s, provider=%s,%+v", pubmsg.From, deal.Proposal.Provider, pubAddrs)
 	}
 
 	if pubmsg.To != miner2.StorageMarketActorAddr {

markets/storageadapter/dealpublisher.go

@@ -7,27 +7,33 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ipfs/go-cid"
 	"go.uber.org/fx"
+	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
-	"github.com/filecoin-project/lotus/node/config"
+	"github.com/filecoin-project/go-state-types/big"
+	market2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/market"
 
-	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/lotus/api"
-
 	"github.com/filecoin-project/lotus/chain/actors"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/market"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 	"github.com/filecoin-project/lotus/chain/types"
-	market2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/market"
-	"github.com/ipfs/go-cid"
-	"golang.org/x/xerrors"
+	"github.com/filecoin-project/lotus/node/config"
+	"github.com/filecoin-project/lotus/storage"
 )
 
 type dealPublisherAPI interface {
 	ChainHead(context.Context) (*types.TipSet, error)
 	MpoolPushMessage(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec) (*types.SignedMessage, error)
 	StateMinerInfo(context.Context, address.Address, types.TipSetKey) (miner.MinerInfo, error)
+
+	WalletBalance(context.Context, address.Address) (types.BigInt, error)
+	WalletHas(context.Context, address.Address) (bool, error)
+	StateAccountKey(context.Context, address.Address, types.TipSetKey) (address.Address, error)
+	StateLookupID(context.Context, address.Address, types.TipSetKey) (address.Address, error)
 }
 
 // DealPublisher batches deal publishing so that many deals can be included in
@@ -40,6 +46,7 @@ type dealPublisherAPI interface {
 // publish message with all deals in the queue.
 type DealPublisher struct {
 	api dealPublisherAPI
+	as  *storage.AddressSelector
 
 	ctx      context.Context
 	Shutdown context.CancelFunc
@@ -87,14 +94,14 @@ type PublishMsgConfig struct {
 func NewDealPublisher(
 	feeConfig *config.MinerFeeConfig,
 	publishMsgCfg PublishMsgConfig,
-) func(lc fx.Lifecycle, full api.FullNode) *DealPublisher {
-	return func(lc fx.Lifecycle, full api.FullNode) *DealPublisher {
+) func(lc fx.Lifecycle, full api.FullNode, as *storage.AddressSelector) *DealPublisher {
+	return func(lc fx.Lifecycle, full api.FullNode, as *storage.AddressSelector) *DealPublisher {
 		maxFee := abi.NewTokenAmount(0)
 		if feeConfig != nil {
 			maxFee = abi.TokenAmount(feeConfig.MaxPublishDealsFee)
 		}
 		publishSpec := &api.MessageSendSpec{MaxFee: maxFee}
-		dp := newDealPublisher(full, publishMsgCfg, publishSpec)
+		dp := newDealPublisher(full, as, publishMsgCfg, publishSpec)
 		lc.Append(fx.Hook{
 			OnStop: func(ctx context.Context) error {
 				dp.Shutdown()
@@ -107,12 +114,14 @@ func NewDealPublisher(
 
 func newDealPublisher(
 	dpapi dealPublisherAPI,
+	as *storage.AddressSelector,
 	publishMsgCfg PublishMsgConfig,
 	publishSpec *api.MessageSendSpec,
 ) *DealPublisher {
 	ctx, cancel := context.WithCancel(context.Background())
 	return &DealPublisher{
 		api:                   dpapi,
+		as:                    as,
 		ctx:                   ctx,
 		Shutdown:              cancel,
 		maxDealsPerPublishMsg: publishMsgCfg.MaxDealsPerMsg,
@@ -345,9 +354,14 @@ func (p *DealPublisher) publishDealProposals(deals []market2.ClientDealProposal)
 		return cid.Undef, xerrors.Errorf("serializing PublishStorageDeals params failed: %w", err)
 	}
 
+	addr, _, err := p.as.AddressFor(p.ctx, p.api, mi, api.DealPublishAddr, big.Zero(), big.Zero())
+	if err != nil {
+		return cid.Undef, xerrors.Errorf("selecting address for publishing deals: %w", err)
+	}
+
 	smsg, err := p.api.MpoolPushMessage(p.ctx, &types.Message{
 		To:     market.Address,
-		From:   mi.Worker,
+		From:   addr,
 		Value:  types.NewInt(0),
 		Method: market.Methods.PublishStorageDeals,
 		Params: params,