feat: manual publish storage deal message

gql/resolver_dealpublish.go

@@ -8,7 +8,9 @@ import (
 
 	"github.com/filecoin-project/boost/gql/types"
 	cborutil "github.com/filecoin-project/go-cbor-util"
+	"github.com/google/uuid"
 	"github.com/graph-gophers/graphql-go"
+	"github.com/ipfs/go-cid"
 )
 
 // basicDealResolver just has simple types (as opposed to dealResolver which
@@ -32,6 +34,7 @@ type basicDealResolver struct {
 }
 
 type dealPublishResolver struct {
+	ManualPSD      bool
 	Start          graphql.Time
 	Period         int32
 	MaxDealsPerMsg int32
@@ -151,15 +154,47 @@ func (r *resolver) DealPublish(ctx context.Context) (*dealPublishResolver, error
 	}
 
 	return &dealPublishResolver{
+		ManualPSD:      r.publisher.ManualPSD(),
 		Deals:          basicDeals,
 		Period:         int32(pending.PublishPeriod.Seconds()),
 		Start:          graphql.Time{Time: pending.PublishPeriodStart},
 		MaxDealsPerMsg: int32(r.cfg.LotusDealmaking.MaxDealsPerPublishMsg),
 	}, nil
 }
 
-// mutation: dealPublishNow(): bool
 func (r *resolver) DealPublishNow(ctx context.Context) (bool, error) {
 	r.publisher.ForcePublishPendingDeals()
 	return true, nil
 }
+
+// mutation: publishPendingDeals([ID!]!): [ID!]!
+func (r *resolver) PublishPendingDeals(ctx context.Context, args struct{ IDs []graphql.ID }) ([]graphql.ID, error) {
+	var pcids []cid.Cid
+	uuidToPcid := make(map[cid.Cid]uuid.UUID)
+	var ret []graphql.ID
+
+	for _, id := range args.IDs {
+		dealId, err := toUuid(id)
+		if err != nil {
+			return nil, err
+		}
+		deal, err := r.dealsDB.ByID(ctx, dealId)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get deal details from DB %s: %w", dealId.String(), err)
+		}
+		signedProp, err := cborutil.AsIpld(&deal.ClientDealProposal)
+		if err != nil {
+			return nil, fmt.Errorf("error in generating proposal cid for deal %s: %w", dealId.String(), err)
+		}
+		pcid := signedProp.Cid()
+		uuidToPcid[pcid] = dealId
+		pcids = append(pcids, pcid)
+	}
+
+	publishedCids := r.publisher.PublishQueuedDeals(pcids)
+	for _, c := range publishedCids {
+		ret = append(ret, graphql.ID(uuidToPcid[c].String()))
+	}
+
+	return ret, nil
+}

gql/schema.graphql

@@ -412,6 +412,7 @@ type FundsLog {
 }
 
 type DealPublish {
+  ManualPSD: Boolean!
   Period: Int!
   Start: Time!
   MaxDealsPerMsg: Int!
@@ -609,6 +610,9 @@ type RootMutation {
 
   """Update the Storage Ask (price of doing a storage deal)"""
   storageAskUpdate(update: StorageAskUpdate!): Boolean!
+
+  """Publish the deal for the supplied deal UUIDs"""
+  publishPendingDeals(ids: [ID!]!): [ID!]!
 }
 
 type RootSubscription {

markets/storageadapter/dealpublisher.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	cborutil "github.com/filecoin-project/go-cbor-util"
 	"github.com/ipfs/go-cid"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
@@ -53,6 +54,7 @@ type DealPublisher struct {
 	ctx      context.Context
 	Shutdown context.CancelFunc
 
+	manualPSD             bool
 	maxDealsPerPublishMsg uint64
 	publishPeriod         time.Duration
 	publishSpec           *api.MessageSendSpec
@@ -66,9 +68,10 @@ type DealPublisher struct {
 
 // A deal that is queued to be published
 type pendingDeal struct {
-	ctx    context.Context
-	deal   market.ClientDealProposal
-	Result chan publishResult
+	ctx         context.Context
+	deal        market.ClientDealProposal
+	Result      chan publishResult
+	proposalCid cid.Cid
 }
 
 // The result of publishing a deal
@@ -77,12 +80,20 @@ type publishResult struct {
 	err    error
 }
 
-func newPendingDeal(ctx context.Context, deal market.ClientDealProposal) *pendingDeal {
-	return &pendingDeal{
-		ctx:    ctx,
-		deal:   deal,
-		Result: make(chan publishResult),
+func newPendingDeal(ctx context.Context, deal market.ClientDealProposal) (*pendingDeal, error) {
+
+	// Generate unique identifier for the deal
+	signedProp, err := cborutil.AsIpld(&deal)
+	if err != nil {
+		return nil, fmt.Errorf("failed to compute signed deal proposal ipld node: %w", err)
 	}
+
+	return &pendingDeal{
+		ctx:         ctx,
+		deal:        deal,
+		Result:      make(chan publishResult),
+		proposalCid: signedProp.Cid(),
+	}, nil
 }
 
 type PublishMsgConfig struct {
@@ -94,6 +105,9 @@ type PublishMsgConfig struct {
 	MaxDealsPerMsg uint64
 	// Minimum start epoch buffer to give time for sealing of sector with deal
 	StartEpochSealingBuffer uint64
+	// Enable ExternalDealPublishControl to allow users to choose when to send Publish messages
+	// for each deal
+	ExternalDealPublishControl bool
 }
 
 func NewDealPublisher(
@@ -133,6 +147,7 @@ func newDealPublisher(
 		publishPeriod:           publishMsgCfg.Period,
 		startEpochSealingBuffer: abi.ChainEpoch(publishMsgCfg.StartEpochSealingBuffer),
 		publishSpec:             publishSpec,
+		manualPSD:               publishMsgCfg.ExternalDealPublishControl,
 	}
 }
 
@@ -172,7 +187,10 @@ func (p *DealPublisher) ForcePublishPendingDeals() {
 }
 
 func (p *DealPublisher) Publish(ctx context.Context, deal market.ClientDealProposal) (cid.Cid, error) {
-	pdeal := newPendingDeal(ctx, deal)
+	pdeal, err := newPendingDeal(ctx, deal)
+	if err != nil {
+		return cid.Undef, fmt.Errorf("failed create pending deal: %w", err)
+	}
 
 	// Add the deal to the queue
 	p.processNewDeal(pdeal)
@@ -209,6 +227,11 @@ func (p *DealPublisher) processNewDeal(pdeal *pendingDeal) {
 	log.Infof("add deal with piece CID %s to publish deals queue - %d deals in queue (max queue size %d)",
 		pdeal.deal.Proposal.PieceCID, len(p.pending), p.maxDealsPerPublishMsg)
 
+	// Return from here if manual PSD is enabled
+	if p.manualPSD {
+		return
+	}
+
 	// If the maximum number of deals per message has been reached or we're not batching, send a
 	// publish message
 	if uint64(len(p.pending)) >= p.maxDealsPerPublishMsg || p.publishPeriod == 0 {
@@ -257,6 +280,7 @@ func (p *DealPublisher) waitForMoreDeals() {
 }
 
 func (p *DealPublisher) publishAllDeals() {
+
 	// If the timeout hasn't yet been cancelled, cancel it
 	if p.cancelWaitForMoreDeals != nil {
 		p.cancelWaitForMoreDeals()
@@ -443,3 +467,63 @@ func (p *DealPublisher) filterCancelledDeals() {
 	}
 	p.pending = filtered
 }
+
+func (p *DealPublisher) PublishQueuedDeals(deals []cid.Cid) []cid.Cid {
+	p.lk.Lock()
+	var ret []cid.Cid
+	var toPublish []*pendingDeal
+
+	// Check that each deal is part of the queue
+	for _, c := range deals {
+		found := false
+		for _, pd := range p.pending {
+			if c.Equals(pd.proposalCid) {
+				found = true
+				toPublish = append(toPublish, pd)
+				break
+			}
+		}
+		if found {
+			ret = append(ret, c)
+		} else {
+			log.Debugf("failed to find the proposal %S in pending deals", c)
+		}
+	}
+
+	p.lk.Unlock()
+
+	log.Infof("publishing deal proposals: %s", ret)
+
+	// Remove the deal from Pending
+	p.cleanupPending(deals)
+
+	// Send the publish message
+	go p.publishReady(toPublish)
+	return ret
+}
+
+func (p *DealPublisher) cleanupPending(dealIds []cid.Cid) {
+	p.lk.Lock()
+	defer p.lk.Unlock()
+
+	var newPending []*pendingDeal
+
+	for _, dp := range p.pending {
+		found := false
+		for _, id := range dealIds {
+			if id == dp.proposalCid {
+				found = true
+			}
+		}
+		if !found {
+			newPending = append(newPending, dp)
+		}
+	}
+
+	p.pending = newPending
+
+}
+
+func (p *DealPublisher) ManualPSD() bool {
+	return p.manualPSD
+}

markets/storageadapter/dealpublisher_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	cborutil "github.com/filecoin-project/go-cbor-util"
 	"github.com/ipfs/go-cid"
 	"github.com/raulk/clock"
 	"github.com/stretchr/testify/require"
@@ -232,6 +233,61 @@ func TestForcePublish(t *testing.T) {
 	checkPublishedDeals(t, dpapi, dealsToPublish, []int{2})
 }
 
+func TestPublishPendingDeals(t *testing.T) {
+	//stm: @MARKET_DEAL_PUBLISHER_PUBLISH_001, @MARKET_DEAL_PUBLISHER_GET_PENDING_DEALS_001
+	//stm: @MARKET_DEAL_PUBLISHER_FORCE_PUBLISH_ALL_001
+	dpapi := newDPAPI(t)
+
+	// Create a deal publisher
+	start := build.Clock.Now()
+	publishPeriod := time.Hour
+	dp := newDealPublisher(dpapi, nil, PublishMsgConfig{
+		Period:         publishPeriod,
+		MaxDealsPerMsg: 10,
+	}, &api.MessageSendSpec{MaxFee: abi.NewTokenAmount(1)})
+
+	// Queue three deals for publishing, one with a cancelled context
+	var dealsToPublish []markettypes.ClientDealProposal
+	// 1. Regular deal
+	deal := publishDeal(t, dp, 0, false, false)
+	dealsToPublish = append(dealsToPublish, deal)
+	// 2. Deal with cancelled context
+	publishDeal(t, dp, 0, true, false)
+	// 3. Regular deal
+	deal = publishDeal(t, dp, 0, false, false)
+	dealsToPublish = append(dealsToPublish, deal)
+
+	// Allow a moment for them to be queued
+	build.Clock.Sleep(10 * time.Millisecond)
+
+	// Should be two deals in the pending deals list
+	// (deal with cancelled context is ignored)
+	pendingInfo := dp.PendingDeals()
+	require.Len(t, pendingInfo.Deals, 2)
+	require.Equal(t, publishPeriod, pendingInfo.PublishPeriod)
+	require.True(t, pendingInfo.PublishPeriodStart.After(start))
+	require.True(t, pendingInfo.PublishPeriodStart.Before(build.Clock.Now()))
+
+	var pcids []cid.Cid
+	props := pendingInfo.Deals
+	for _, p := range props {
+		signedProp, err := cborutil.AsIpld(&p)
+		require.NoError(t, err)
+		pcids = append(pcids, signedProp.Cid())
+	}
+
+	// Publish all pending deals and verify all have been published
+	publishedDeals := dp.PublishQueuedDeals(pcids)
+	require.Equal(t, pcids, publishedDeals)
+
+	// Should be no pending deals
+	pendingInfo = dp.PendingDeals()
+	require.Len(t, pendingInfo.Deals, 0)
+
+	// Make sure the expected deals were published
+	checkPublishedDeals(t, dpapi, dealsToPublish, []int{2})
+}
+
 func publishDeal(t *testing.T, dp *DealPublisher, invalid int, ctxCancelled bool, expired bool) markettypes.ClientDealProposal {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)

node/builder.go

@@ -626,9 +626,10 @@ func ConfigBoost(cfg *config.Boost) Option {
 		),
 
 		Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(&legacyFees, storageadapter.PublishMsgConfig{
-			Period:                  time.Duration(cfg.LotusDealmaking.PublishMsgPeriod),
-			MaxDealsPerMsg:          cfg.LotusDealmaking.MaxDealsPerPublishMsg,
-			StartEpochSealingBuffer: cfg.LotusDealmaking.StartEpochSealingBuffer,
+			Period:                     time.Duration(cfg.LotusDealmaking.PublishMsgPeriod),
+			MaxDealsPerMsg:             cfg.LotusDealmaking.MaxDealsPerPublishMsg,
+			StartEpochSealingBuffer:    cfg.LotusDealmaking.StartEpochSealingBuffer,
+			ExternalDealPublishControl: cfg.Dealmaking.ManualDealPublish,
 		})),
 
 		Override(new(sealer.Unsealer), From(new(lotus_modules.MinerStorageService))),

node/config/def.go

@@ -142,6 +142,7 @@ func DefaultBoost() *Boost {
 			SealingPipelineCacheTimeout:        Duration(30 * time.Second),
 			FundsTaggingEnabled:                true,
 			EnableLegacyStorageDeals:           false,
+			ManualDealPublish:                  false,
 		},
 
 		LotusDealmaking: lotus_config.DealmakingConfig{

node/config/types.go

@@ -272,6 +272,11 @@ type DealmakingConfig struct {
 	// Whether to enable legacy deals on the Boost node or not. We recommend keeping
 	// them disabled. These will be completely deprecated soon.
 	EnableLegacyStorageDeals bool
+
+	// When set to true, the user is responsible for publishing deals manually.
+	// The values of MaxDealsPerPublishMsg and PublishMsgPeriod will be
+	// ignored, and deals will remain in the pending state until manually published.
+	ManualDealPublish bool
 }
 
 type ContractDealsConfig struct {