Merge pull request #4730 from filecoin-project/feat/deal-id-change-in-sector-committed

Handle deal id changes in OnDealSectorCommitted

go.mod

@@ -28,9 +28,9 @@ require (
 	github.com/filecoin-project/go-bitfield v0.2.2
 	github.com/filecoin-project/go-cbor-util v0.0.0-20191219014500-08c40a1e63a2
 	github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03
-	github.com/filecoin-project/go-data-transfer v1.0.0
+	github.com/filecoin-project/go-data-transfer v1.0.1
 	github.com/filecoin-project/go-fil-commcid v0.0.0-20200716160307-8f644712406f
-	github.com/filecoin-project/go-fil-markets v1.0.0
+	github.com/filecoin-project/go-fil-markets v1.0.1
 	github.com/filecoin-project/go-jsonrpc v0.1.2-0.20201008195726-68c6a2704e49
 	github.com/filecoin-project/go-multistore v0.0.3
 	github.com/filecoin-project/go-padreader v0.0.0-20200903213702-ed5fae088b20
@@ -135,6 +135,7 @@ require (
 	go.uber.org/fx v1.9.0
 	go.uber.org/multierr v1.5.0
 	go.uber.org/zap v1.15.0
+	golang.org/x/net v0.0.0-20200707034311-ab3426394381
 	golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208
 	golang.org/x/sys v0.0.0-20200926100807-9d91bd62050c
 	golang.org/x/time v0.0.0-20191024005414-555d28b269f0

go.sum

@@ -247,16 +247,14 @@ github.com/filecoin-project/go-cbor-util v0.0.0-20191219014500-08c40a1e63a2 h1:a
 github.com/filecoin-project/go-cbor-util v0.0.0-20191219014500-08c40a1e63a2/go.mod h1:pqTiPHobNkOVM5thSRsHYjyQfq7O5QSCMhvuu9JoDlg=
 github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03 h1:2pMXdBnCiXjfCYx/hLqFxccPoqsSveQFxVLvNxy9bus=
 github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03/go.mod h1:+viYnvGtUTgJRdy6oaeF4MTFKAfatX071MPDPBL11EQ=
-github.com/filecoin-project/go-data-transfer v0.9.0 h1:nTT8j7Hu3TM0wRWrGy83/ctawG7sleJGdFWtIsUsKgY=
+github.com/filecoin-project/go-data-transfer v1.0.1 h1:5sYKDbstyDsdJpVP4UGUW6+BgCNfgnH8hQgf0E3ZAno=
-github.com/filecoin-project/go-data-transfer v0.9.0/go.mod h1:i2CqUy7TMQGKukj9BgqIxiP8nDHDXU2VLd771KVaCaQ=
+github.com/filecoin-project/go-data-transfer v1.0.1/go.mod h1:UxvfUAY9v3ub0a21BSK9u3pB2aq30Y0KMsG+w9/ysyo=
-github.com/filecoin-project/go-data-transfer v1.0.0 h1:spY2EfWmf03dYNI24tmbHThXGbXlcVHF65ho9x0iN9A=
-github.com/filecoin-project/go-data-transfer v1.0.0/go.mod h1:fT8+EXD0Ghy0S9tulpMpFOyBquVjsNICxLTLeSEV+mw=
 github.com/filecoin-project/go-ds-versioning v0.1.0 h1:y/X6UksYTsK8TLCI7rttCKEvl8btmWxyFMEeeWGUxIQ=
 github.com/filecoin-project/go-ds-versioning v0.1.0/go.mod h1:mp16rb4i2QPmxBnmanUx8i/XANp+PFCCJWiAb+VW4/s=
 github.com/filecoin-project/go-fil-commcid v0.0.0-20200716160307-8f644712406f h1:GxJzR3oRIMTPtpZ0b7QF8FKPK6/iPAc7trhlL5k/g+s=
 github.com/filecoin-project/go-fil-commcid v0.0.0-20200716160307-8f644712406f/go.mod h1:Eaox7Hvus1JgPrL5+M3+h7aSPHc0cVqpSxA+TxIEpZQ=
-github.com/filecoin-project/go-fil-markets v1.0.0 h1:np9+tlnWXh9xYG4oZfha6HZFLYOaAZoMGR3V4w6DM48=
+github.com/filecoin-project/go-fil-markets v1.0.1 h1:xyhsHPnaECkOP8amFlp/2nK1L5xaQ9mCscXXxX2lwcs=
-github.com/filecoin-project/go-fil-markets v1.0.0/go.mod h1:lXExJyYHwpMMddCqhEdNrc7euYJKNkp04K76NZqJLGg=
+github.com/filecoin-project/go-fil-markets v1.0.1/go.mod h1:qdAqt05NWpmkGycb4duXaMtidLDmyaz1aG5goWIMm/E=
 github.com/filecoin-project/go-hamt-ipld v0.1.5 h1:uoXrKbCQZ49OHpsTCkrThPNelC4W3LPEk0OrS/ytIBM=
 github.com/filecoin-project/go-hamt-ipld v0.1.5 h1:uoXrKbCQZ49OHpsTCkrThPNelC4W3LPEk0OrS/ytIBM=
 github.com/filecoin-project/go-hamt-ipld v0.1.5/go.mod h1:6Is+ONR5Cd5R6XZoCse1CWaXZc0Hdb/JeX+EQCQzX24=
@@ -549,10 +547,6 @@ github.com/ipfs/go-filestore v1.0.0/go.mod h1:/XOCuNtIe2f1YPbiXdYvD0BKLA0JR1MgPi
 github.com/ipfs/go-fs-lock v0.0.6 h1:sn3TWwNVQqSeNjlWy6zQ1uUGAZrV3hPOyEA6y1/N2a0=
 github.com/ipfs/go-fs-lock v0.0.6/go.mod h1:OTR+Rj9sHiRubJh3dRhD15Juhd/+w6VPOY28L7zESmM=
 github.com/ipfs/go-graphsync v0.1.0/go.mod h1:jMXfqIEDFukLPZHqDPp8tJMbHO9Rmeb9CEGevngQbmE=
-github.com/ipfs/go-graphsync v0.3.0 h1:I6Y20kSuCWkUvPoUWo4V3am704/9QjgDVVkf0zIV8+8=
-github.com/ipfs/go-graphsync v0.3.0/go.mod h1:gEBvJUNelzMkaRPJTpg/jaKN4AQW/7wDWu0K92D8o10=
-github.com/ipfs/go-graphsync v0.3.1 h1:dJLYrck4oyJDfMVhGEKiWHxaY8oYMWko4m2Fi+4bofo=
-github.com/ipfs/go-graphsync v0.3.1/go.mod h1:bw4LiLM5Oq/uLdzEtih9LK8GrwSijv+XqYiWCTxHMqs=
 github.com/ipfs/go-graphsync v0.4.2 h1:Y/jt5r619yj0LI7OLtGKh4jYm8goYUcuJ09y7TZ3zMo=
 github.com/ipfs/go-graphsync v0.4.2/go.mod h1:/VmbZTUdUMTbNkgzAiCEucIIAU3BkLE2cZrDCVUhyi0=
 github.com/ipfs/go-hamt-ipld v0.1.1 h1:0IQdvwnAAUKmDE+PMJa5y1QiwOPHpI9+eAbQEEEYthk=

markets/storageadapter/client.go

@@ -23,7 +23,7 @@ import (
 
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
-	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	marketactor "github.com/filecoin-project/lotus/chain/actors/builtin/market"
 	"github.com/filecoin-project/lotus/chain/events"
 	"github.com/filecoin-project/lotus/chain/events/state"
 	"github.com/filecoin-project/lotus/chain/market"
@@ -213,105 +213,8 @@ func (c *ClientNodeAdapter) DealProviderCollateralBounds(ctx context.Context, si
 	return big.Mul(bounds.Min, big.NewInt(clientOverestimation)), bounds.Max, nil
 }
 
-func (c *ClientNodeAdapter) OnDealSectorCommitted(ctx context.Context, provider address.Address, dealId abi.DealID, cb storagemarket.DealSectorCommittedCallback) error {
+func (c *ClientNodeAdapter) OnDealSectorCommitted(ctx context.Context, provider address.Address, dealID abi.DealID, proposal market2.DealProposal, publishCid *cid.Cid, cb storagemarket.DealSectorCommittedCallback) error {
-	checkFunc := func(ts *types.TipSet) (done bool, more bool, err error) {
+	return OnDealSectorCommitted(ctx, c, c.ev, provider, dealID, marketactor.DealProposal(proposal), publishCid, cb)
-		sd, err := c.StateMarketStorageDeal(ctx, dealId, ts.Key())
-
-		if err != nil {
-			// TODO: This may be fine for some errors
-			return false, false, xerrors.Errorf("client: failed to look up deal on chain: %w", err)
-		}
-
-		if sd.State.SectorStartEpoch > 0 {
-			cb(nil)
-			return true, false, nil
-		}
-
-		return false, true, nil
-	}
-
-	called := func(msg *types.Message, rec *types.MessageReceipt, ts *types.TipSet, curH abi.ChainEpoch) (more bool, err error) {
-		defer func() {
-			if err != nil {
-				cb(xerrors.Errorf("handling applied event: %w", err))
-			}
-		}()
-
-		if msg == nil {
-			log.Error("timed out waiting for deal activation... what now?")
-			return false, nil
-		}
-
-		sd, err := c.StateMarketStorageDeal(ctx, dealId, ts.Key())
-		if err != nil {
-			return false, xerrors.Errorf("failed to look up deal on chain: %w", err)
-		}
-
-		if sd.State.SectorStartEpoch < 1 {
-			return false, xerrors.Errorf("deal wasn't active: deal=%d, parentState=%s, h=%d", dealId, ts.ParentState(), ts.Height())
-		}
-
-		log.Infof("Storage deal %d activated at epoch %d", dealId, sd.State.SectorStartEpoch)
-
-		cb(nil)
-
-		return false, nil
-	}
-
-	revert := func(ctx context.Context, ts *types.TipSet) error {
-		log.Warn("deal activation reverted; TODO: actually handle this!")
-		// TODO: Just go back to DealSealing?
-		return nil
-	}
-
-	var sectorNumber abi.SectorNumber
-	var sectorFound bool
-	matchEvent := func(msg *types.Message) (matched bool, err error) {
-		if msg.To != provider {
-			return false, nil
-		}
-
-		switch msg.Method {
-		case miner2.MethodsMiner.PreCommitSector:
-			var params miner.SectorPreCommitInfo
-			if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
-				return false, xerrors.Errorf("unmarshal pre commit: %w", err)
-			}
-
-			for _, did := range params.DealIDs {
-				if did == dealId {
-					sectorNumber = params.SectorNumber
-					sectorFound = true
-					return false, nil
-				}
-			}
-
-			return false, nil
-		case miner2.MethodsMiner.ProveCommitSector:
-			var params miner.ProveCommitSectorParams
-			if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
-				return false, xerrors.Errorf("failed to unmarshal prove commit sector params: %w", err)
-			}
-
-			if !sectorFound {
-				return false, nil
-			}
-
-			if params.SectorNumber != sectorNumber {
-				return false, nil
-			}
-
-			return true, nil
-		default:
-			return false, nil
-		}
-	}
-
-	if err := c.ev.Called(checkFunc, called, revert, int(build.MessageConfidence+1), events.NoTimeout, matchEvent); err != nil {
-		return xerrors.Errorf("failed to set up called handler: %w", err)
-	}
-
-	return nil
 }
 
 func (c *ClientNodeAdapter) OnDealExpiredOrSlashed(ctx context.Context, dealID abi.DealID, onDealExpired storagemarket.DealExpiredCallback, onDealSlashed storagemarket.DealSlashedCallback) error {

markets/storageadapter/getcurrentdealinfo.go

@@ -0,0 +1,99 @@
+package storageadapter
+
+import (
+	"bytes"
+	"context"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/exitcode"
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/market"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/ipfs/go-cid"
+	"golang.org/x/xerrors"
+)
+
+type getCurrentDealInfoAPI interface {
+	StateLookupID(context.Context, address.Address, types.TipSetKey) (address.Address, error)
+	StateMarketStorageDeal(context.Context, abi.DealID, types.TipSetKey) (*api.MarketDeal, error)
+	StateSearchMsg(context.Context, cid.Cid) (*api.MsgLookup, error)
+}
+
+// GetCurrentDealInfo gets current information on a deal, and corrects the deal ID as needed
+func GetCurrentDealInfo(ctx context.Context, ts *types.TipSet, api getCurrentDealInfoAPI, dealID abi.DealID, proposal market.DealProposal, publishCid *cid.Cid) (abi.DealID, *api.MarketDeal, error) {
+	marketDeal, dealErr := api.StateMarketStorageDeal(ctx, dealID, ts.Key())
+	if dealErr == nil {
+		equal, err := checkDealEquality(ctx, ts, api, proposal, marketDeal.Proposal)
+		if err != nil {
+			return dealID, nil, err
+		}
+		if equal {
+			return dealID, marketDeal, nil
+		}
+		dealErr = xerrors.Errorf("Deal proposals did not match")
+	}
+	if publishCid == nil {
+		return dealID, nil, dealErr
+	}
+	// attempt deal id correction
+	lookup, err := api.StateSearchMsg(ctx, *publishCid)
+	if err != nil {
+		return dealID, nil, err
+	}
+
+	if lookup.Receipt.ExitCode != exitcode.Ok {
+		return dealID, nil, xerrors.Errorf("looking for publish deal message %s: non-ok exit code: %s", *publishCid, lookup.Receipt.ExitCode)
+	}
+
+	var retval market.PublishStorageDealsReturn
+	if err := retval.UnmarshalCBOR(bytes.NewReader(lookup.Receipt.Return)); err != nil {
+		return dealID, nil, xerrors.Errorf("looking for publish deal message: unmarshaling message return: %w", err)
+	}
+
+	if len(retval.IDs) != 1 {
+		// market currently only ever sends messages with 1 deal
+		return dealID, nil, xerrors.Errorf("can't recover dealIDs from publish deal message with more than 1 deal")
+	}
+
+	if retval.IDs[0] == dealID {
+		// DealID did not change, so we are stuck with the original lookup error
+		return dealID, nil, dealErr
+	}
+
+	dealID = retval.IDs[0]
+	marketDeal, err = api.StateMarketStorageDeal(ctx, dealID, ts.Key())
+
+	if err == nil {
+		equal, err := checkDealEquality(ctx, ts, api, proposal, marketDeal.Proposal)
+		if err != nil {
+			return dealID, nil, err
+		}
+		if !equal {
+			return dealID, nil, xerrors.Errorf("Deal proposals did not match")
+		}
+	}
+	return dealID, marketDeal, err
+}
+
+func checkDealEquality(ctx context.Context, ts *types.TipSet, api getCurrentDealInfoAPI, p1, p2 market.DealProposal) (bool, error) {
+	p1ClientID, err := api.StateLookupID(ctx, p1.Client, ts.Key())
+	if err != nil {
+		return false, err
+	}
+	p2ClientID, err := api.StateLookupID(ctx, p2.Client, ts.Key())
+	if err != nil {
+		return false, err
+	}
+	return p1.PieceCID.Equals(p2.PieceCID) &&
+		p1.PieceSize == p2.PieceSize &&
+		p1.VerifiedDeal == p2.VerifiedDeal &&
+		p1.Label == p2.Label &&
+		p1.StartEpoch == p2.StartEpoch &&
+		p1.EndEpoch == p2.EndEpoch &&
+		p1.StoragePricePerEpoch.Equals(p2.StoragePricePerEpoch) &&
+		p1.ProviderCollateral.Equals(p2.ProviderCollateral) &&
+		p1.ClientCollateral.Equals(p2.ClientCollateral) &&
+		p1.Provider == p2.Provider &&
+		p1ClientID == p2ClientID, nil
+}

markets/storageadapter/getcurrentdealinfo_test.go

@@ -0,0 +1,263 @@
+package storageadapter
+
+import (
+	"bytes"
+	"errors"
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/exitcode"
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/market"
+	test "github.com/filecoin-project/lotus/chain/events/state/mock"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/ipfs/go-cid"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/net/context"
+	"golang.org/x/xerrors"
+)
+
+var errNotFound = errors.New("Could not find")
+
+func TestGetCurrentDealInfo(t *testing.T) {
+	ctx := context.Background()
+	dummyCid, _ := cid.Parse("bafkqaaa")
+	startDealID := abi.DealID(rand.Uint64())
+	newDealID := abi.DealID(rand.Uint64())
+	twoValuesReturn := makePublishDealsReturnBytes(t, []abi.DealID{abi.DealID(rand.Uint64()), abi.DealID(rand.Uint64())})
+	sameValueReturn := makePublishDealsReturnBytes(t, []abi.DealID{startDealID})
+	newValueReturn := makePublishDealsReturnBytes(t, []abi.DealID{newDealID})
+	proposal := market.DealProposal{
+		PieceCID:  dummyCid,
+		PieceSize: abi.PaddedPieceSize(rand.Uint64()),
+		Label:     "success",
+	}
+	otherProposal := market.DealProposal{
+		PieceCID:  dummyCid,
+		PieceSize: abi.PaddedPieceSize(rand.Uint64()),
+		Label:     "other",
+	}
+	successDeal := &api.MarketDeal{
+		Proposal: proposal,
+		State: market.DealState{
+			SectorStartEpoch: 1,
+			LastUpdatedEpoch: 2,
+		},
+	}
+	otherDeal := &api.MarketDeal{
+		Proposal: otherProposal,
+		State: market.DealState{
+			SectorStartEpoch: 1,
+			LastUpdatedEpoch: 2,
+		},
+	}
+	testCases := map[string]struct {
+		searchMessageLookup *api.MsgLookup
+		searchMessageErr    error
+		marketDeals         map[abi.DealID]*api.MarketDeal
+		publishCid          *cid.Cid
+		expectedDealID      abi.DealID
+		expectedMarketDeal  *api.MarketDeal
+		expectedError       error
+	}{
+		"deal lookup succeeds": {
+			marketDeals: map[abi.DealID]*api.MarketDeal{
+				startDealID: successDeal,
+			},
+			expectedDealID:     startDealID,
+			expectedMarketDeal: successDeal,
+		},
+		"publish CID = nil": {
+			expectedDealID: startDealID,
+			expectedError:  errNotFound,
+		},
+		"publish CID = nil, other deal on lookup": {
+			marketDeals: map[abi.DealID]*api.MarketDeal{
+				startDealID: otherDeal,
+			},
+			expectedDealID: startDealID,
+			expectedError:  xerrors.Errorf("Deal proposals did not match"),
+		},
+		"search message fails": {
+			publishCid:       &dummyCid,
+			searchMessageErr: errors.New("something went wrong"),
+			expectedDealID:   startDealID,
+			expectedError:    errors.New("something went wrong"),
+		},
+		"return code not ok": {
+			publishCid: &dummyCid,
+			searchMessageLookup: &api.MsgLookup{
+				Receipt: types.MessageReceipt{
+					ExitCode: exitcode.ErrIllegalState,
+				},
+			},
+			expectedDealID: startDealID,
+			expectedError:  xerrors.Errorf("looking for publish deal message %s: non-ok exit code: %s", dummyCid, exitcode.ErrIllegalState),
+		},
+		"unable to unmarshal params": {
+			publishCid: &dummyCid,
+			searchMessageLookup: &api.MsgLookup{
+				Receipt: types.MessageReceipt{
+					ExitCode: exitcode.Ok,
+					Return:   []byte("applesauce"),
+				},
+			},
+			expectedDealID: startDealID,
+			expectedError:  xerrors.Errorf("looking for publish deal message: unmarshaling message return: cbor input should be of type array"),
+		},
+		"more than one returned id": {
+			publishCid: &dummyCid,
+			searchMessageLookup: &api.MsgLookup{
+				Receipt: types.MessageReceipt{
+					ExitCode: exitcode.Ok,
+					Return:   twoValuesReturn,
+				},
+			},
+			expectedDealID: startDealID,
+			expectedError:  xerrors.Errorf("can't recover dealIDs from publish deal message with more than 1 deal"),
+		},
+		"deal ids still match": {
+			publishCid: &dummyCid,
+			searchMessageLookup: &api.MsgLookup{
+				Receipt: types.MessageReceipt{
+					ExitCode: exitcode.Ok,
+					Return:   sameValueReturn,
+				},
+			},
+			expectedDealID: startDealID,
+			expectedError:  errNotFound,
+		},
+		"new deal id success": {
+			publishCid: &dummyCid,
+			searchMessageLookup: &api.MsgLookup{
+				Receipt: types.MessageReceipt{
+					ExitCode: exitcode.Ok,
+					Return:   newValueReturn,
+				},
+			},
+			marketDeals: map[abi.DealID]*api.MarketDeal{
+				newDealID: successDeal,
+			},
+			expectedDealID:     newDealID,
+			expectedMarketDeal: successDeal,
+		},
+		"new deal id after other deal found": {
+			publishCid: &dummyCid,
+			searchMessageLookup: &api.MsgLookup{
+				Receipt: types.MessageReceipt{
+					ExitCode: exitcode.Ok,
+					Return:   newValueReturn,
+				},
+			},
+			marketDeals: map[abi.DealID]*api.MarketDeal{
+				startDealID: otherDeal,
+				newDealID:   successDeal,
+			},
+			expectedDealID:     newDealID,
+			expectedMarketDeal: successDeal,
+		},
+		"new deal id failure": {
+			publishCid: &dummyCid,
+			searchMessageLookup: &api.MsgLookup{
+				Receipt: types.MessageReceipt{
+					ExitCode: exitcode.Ok,
+					Return:   newValueReturn,
+				},
+			},
+			expectedDealID: newDealID,
+			expectedError:  errNotFound,
+		},
+		"new deal id, failure due to other deal present": {
+			publishCid: &dummyCid,
+			searchMessageLookup: &api.MsgLookup{
+				Receipt: types.MessageReceipt{
+					ExitCode: exitcode.Ok,
+					Return:   newValueReturn,
+				},
+			},
+			marketDeals: map[abi.DealID]*api.MarketDeal{
+				newDealID: otherDeal,
+			},
+			expectedDealID: newDealID,
+			expectedError:  xerrors.Errorf("Deal proposals did not match"),
+		},
+	}
+	runTestCase := func(testCase string, data struct {
+		searchMessageLookup *api.MsgLookup
+		searchMessageErr    error
+		marketDeals         map[abi.DealID]*api.MarketDeal
+		publishCid          *cid.Cid
+		expectedDealID      abi.DealID
+		expectedMarketDeal  *api.MarketDeal
+		expectedError       error
+	}) {
+		t.Run(testCase, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+			defer cancel()
+			ts, err := test.MockTipset(address.TestAddress, rand.Uint64())
+			require.NoError(t, err)
+			marketDeals := make(map[marketDealKey]*api.MarketDeal)
+			for dealID, deal := range data.marketDeals {
+				marketDeals[marketDealKey{dealID, ts.Key()}] = deal
+			}
+			api := &mockGetCurrentDealInfoAPI{
+				SearchMessageLookup: data.searchMessageLookup,
+				SearchMessageErr:    data.searchMessageErr,
+				MarketDeals:         marketDeals,
+			}
+
+			dealID, marketDeal, err := GetCurrentDealInfo(ctx, ts, api, startDealID, proposal, data.publishCid)
+			require.Equal(t, data.expectedDealID, dealID)
+			require.Equal(t, data.expectedMarketDeal, marketDeal)
+			if data.expectedError == nil {
+				require.NoError(t, err)
+			} else {
+				require.EqualError(t, err, data.expectedError.Error())
+			}
+		})
+	}
+	for testCase, data := range testCases {
+		runTestCase(testCase, data)
+	}
+}
+
+type marketDealKey struct {
+	abi.DealID
+	types.TipSetKey
+}
+
+type mockGetCurrentDealInfoAPI struct {
+	SearchMessageLookup *api.MsgLookup
+	SearchMessageErr    error
+
+	MarketDeals map[marketDealKey]*api.MarketDeal
+}
+
+func (mapi *mockGetCurrentDealInfoAPI) StateMarketStorageDeal(ctx context.Context, dealID abi.DealID, ts types.TipSetKey) (*api.MarketDeal, error) {
+	deal, ok := mapi.MarketDeals[marketDealKey{dealID, ts}]
+	if !ok {
+		return nil, errNotFound
+	}
+	return deal, nil
+}
+
+func (mapi *mockGetCurrentDealInfoAPI) StateSearchMsg(context.Context, cid.Cid) (*api.MsgLookup, error) {
+	return mapi.SearchMessageLookup, mapi.SearchMessageErr
+}
+
+func (mapi *mockGetCurrentDealInfoAPI) StateLookupID(ctx context.Context, addr address.Address, ts types.TipSetKey) (address.Address, error) {
+	return addr, nil
+}
+
+func makePublishDealsReturnBytes(t *testing.T, dealIDs []abi.DealID) []byte {
+	buf := new(bytes.Buffer)
+	dealsReturn := market.PublishStorageDealsReturn{
+		IDs: dealIDs,
+	}
+	err := dealsReturn.MarshalCBOR(buf)
+	require.NoError(t, err)
+	return buf.Bytes()
+}

markets/storageadapter/ondealsectorcommitted.go

@@ -0,0 +1,134 @@
+package storageadapter
+
+import (
+	"bytes"
+	"context"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-fil-markets/storagemarket"
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/market"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/chain/events"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/ipfs/go-cid"
+	"golang.org/x/xerrors"
+)
+
+type sectorCommittedEventsAPI interface {
+	Called(check events.CheckFunc, msgHnd events.MsgHandler, rev events.RevertHandler, confidence int, timeout abi.ChainEpoch, mf events.MsgMatchFunc) error
+}
+
+func OnDealSectorCommitted(ctx context.Context, api getCurrentDealInfoAPI, eventsApi sectorCommittedEventsAPI, provider address.Address, dealID abi.DealID, proposal market.DealProposal, publishCid *cid.Cid, cb storagemarket.DealSectorCommittedCallback) error {
+	checkFunc := func(ts *types.TipSet) (done bool, more bool, err error) {
+		newDealID, sd, err := GetCurrentDealInfo(ctx, ts, api, dealID, proposal, publishCid)
+		if err != nil {
+			// TODO: This may be fine for some errors
+			return false, false, xerrors.Errorf("failed to look up deal on chain: %w", err)
+		}
+		dealID = newDealID
+
+		if sd.State.SectorStartEpoch > 0 {
+			cb(nil)
+			return true, false, nil
+		}
+
+		return false, true, nil
+	}
+
+	var sectorNumber abi.SectorNumber
+	var sectorFound bool
+
+	called := func(msg *types.Message, rec *types.MessageReceipt, ts *types.TipSet, curH abi.ChainEpoch) (more bool, err error) {
+		defer func() {
+			if err != nil {
+				cb(xerrors.Errorf("handling applied event: %w", err))
+			}
+		}()
+		switch msg.Method {
+		case miner.Methods.PreCommitSector:
+			var params miner.SectorPreCommitInfo
+			if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
+				return false, xerrors.Errorf("unmarshal pre commit: %w", err)
+			}
+
+			dealID, _, err = GetCurrentDealInfo(ctx, ts, api, dealID, proposal, publishCid)
+			if err != nil {
+				return false, err
+			}
+
+			for _, did := range params.DealIDs {
+				if did == dealID {
+					sectorNumber = params.SectorNumber
+					sectorFound = true
+					return true, nil
+				}
+			}
+			return true, nil
+		case miner.Methods.ProveCommitSector:
+			if msg == nil {
+				log.Error("timed out waiting for deal activation... what now?")
+				return false, nil
+			}
+
+			_, sd, err := GetCurrentDealInfo(ctx, ts, api, dealID, proposal, publishCid)
+			if err != nil {
+				return false, xerrors.Errorf("failed to look up deal on chain: %w", err)
+			}
+
+			if sd.State.SectorStartEpoch < 1 {
+				return false, xerrors.Errorf("deal wasn't active: deal=%d, parentState=%s, h=%d", dealID, ts.ParentState(), ts.Height())
+			}
+
+			log.Infof("Storage deal %d activated at epoch %d", dealID, sd.State.SectorStartEpoch)
+
+			cb(nil)
+
+			return false, nil
+		default:
+			return false, nil
+		}
+	}
+
+	revert := func(ctx context.Context, ts *types.TipSet) error {
+		log.Warn("deal activation reverted; TODO: actually handle this!")
+		// TODO: Just go back to DealSealing?
+		return nil
+	}
+
+	matchEvent := func(msg *types.Message) (matched bool, err error) {
+		if msg.To != provider {
+			return false, nil
+		}
+
+		switch msg.Method {
+		case miner.Methods.PreCommitSector:
+			return !sectorFound, nil
+		case miner.Methods.ProveCommitSector:
+			if !sectorFound {
+				return false, nil
+			}
+
+			var params miner.ProveCommitSectorParams
+			if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
+				return false, xerrors.Errorf("failed to unmarshal prove commit sector params: %w", err)
+			}
+
+			if params.SectorNumber != sectorNumber {
+				return false, nil
+			}
+
+			return true, nil
+		default:
+			return false, nil
+		}
+
+	}
+
+	if err := eventsApi.Called(checkFunc, called, revert, int(build.MessageConfidence+1), events.NoTimeout, matchEvent); err != nil {
+		return xerrors.Errorf("failed to set up called handler: %w", err)
+	}
+
+	return nil
+}

markets/storageadapter/ondealsectorcommitted_test.go

@@ -0,0 +1,372 @@
+package storageadapter
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"math/rand"
+	"testing"
+
+	blocks "github.com/ipfs/go-block-format"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/cbor"
+	"github.com/filecoin-project/go-state-types/exitcode"
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/market"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/chain/events"
+	test "github.com/filecoin-project/lotus/chain/events/state/mock"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/ipfs/go-cid"
+	"github.com/stretchr/testify/require"
+)
+
+func TestOnDealSectorCommitted(t *testing.T) {
+	provider := address.TestAddress
+	ctx := context.Background()
+	publishCid := generateCids(1)[0]
+	sealedCid := generateCids(1)[0]
+	pieceCid := generateCids(1)[0]
+	startDealID := abi.DealID(rand.Uint64())
+	newDealID := abi.DealID(rand.Uint64())
+	newValueReturn := makePublishDealsReturnBytes(t, []abi.DealID{newDealID})
+	sectorNumber := abi.SectorNumber(rand.Uint64())
+	proposal := market.DealProposal{
+		PieceCID:  pieceCid,
+		PieceSize: abi.PaddedPieceSize(rand.Uint64()),
+		Label:     "success",
+	}
+	unfinishedDeal := &api.MarketDeal{
+		Proposal: proposal,
+		State: market.DealState{
+			SectorStartEpoch: -1,
+			LastUpdatedEpoch: 2,
+		},
+	}
+	successDeal := &api.MarketDeal{
+		Proposal: proposal,
+		State: market.DealState{
+			SectorStartEpoch: 1,
+			LastUpdatedEpoch: 2,
+		},
+	}
+	testCases := map[string]struct {
+		searchMessageLookup *api.MsgLookup
+		searchMessageErr    error
+		checkTsDeals        map[abi.DealID]*api.MarketDeal
+		matchStates         []matchState
+		expectedCBCallCount uint64
+		expectedCBError     error
+		expectedError       error
+	}{
+		"normal sequence": {
+			checkTsDeals: map[abi.DealID]*api.MarketDeal{
+				startDealID: unfinishedDeal,
+			},
+			matchStates: []matchState{
+				{
+					msg: makeMessage(t, provider, miner.Methods.PreCommitSector, &miner.SectorPreCommitInfo{
+						SectorNumber: sectorNumber,
+						SealedCID:    sealedCid,
+						DealIDs:      []abi.DealID{startDealID},
+					}),
+					deals: map[abi.DealID]*api.MarketDeal{
+						startDealID: unfinishedDeal,
+					},
+				},
+				{
+					msg: makeMessage(t, provider, miner.Methods.ProveCommitSector, &miner.ProveCommitSectorParams{
+						SectorNumber: sectorNumber,
+					}),
+					deals: map[abi.DealID]*api.MarketDeal{
+						startDealID: successDeal,
+					},
+				},
+			},
+			expectedCBCallCount: 1,
+		},
+		"deal id changes in called": {
+			searchMessageLookup: &api.MsgLookup{
+				Receipt: types.MessageReceipt{
+					ExitCode: exitcode.Ok,
+					Return:   newValueReturn,
+				},
+			},
+			checkTsDeals: map[abi.DealID]*api.MarketDeal{
+				newDealID: unfinishedDeal,
+			},
+			matchStates: []matchState{
+				{
+					msg: makeMessage(t, provider, miner.Methods.PreCommitSector, &miner.SectorPreCommitInfo{
+						SectorNumber: sectorNumber,
+						SealedCID:    sealedCid,
+						DealIDs:      []abi.DealID{newDealID},
+					}),
+					deals: map[abi.DealID]*api.MarketDeal{
+						newDealID: unfinishedDeal,
+					},
+				},
+				{
+					msg: makeMessage(t, provider, miner.Methods.ProveCommitSector, &miner.ProveCommitSectorParams{
+						SectorNumber: sectorNumber,
+					}),
+					deals: map[abi.DealID]*api.MarketDeal{
+						newDealID: successDeal,
+					},
+				},
+			},
+			expectedCBCallCount: 1,
+		},
+		"deal id changes in precommit": {
+			searchMessageLookup: &api.MsgLookup{
+				Receipt: types.MessageReceipt{
+					ExitCode: exitcode.Ok,
+					Return:   newValueReturn,
+				},
+			},
+			checkTsDeals: map[abi.DealID]*api.MarketDeal{
+				startDealID: unfinishedDeal,
+			},
+			matchStates: []matchState{
+				{
+					msg: makeMessage(t, provider, miner.Methods.PreCommitSector, &miner.SectorPreCommitInfo{
+						SectorNumber: sectorNumber,
+						SealedCID:    sealedCid,
+						DealIDs:      []abi.DealID{newDealID},
+					}),
+					deals: map[abi.DealID]*api.MarketDeal{
+						newDealID: unfinishedDeal,
+					},
+				},
+				{
+					msg: makeMessage(t, provider, miner.Methods.ProveCommitSector, &miner.ProveCommitSectorParams{
+						SectorNumber: sectorNumber,
+					}),
+					deals: map[abi.DealID]*api.MarketDeal{
+						newDealID: successDeal,
+					},
+				},
+			},
+			expectedCBCallCount: 1,
+		},
+		"deal id changes in prove-commit": {
+			searchMessageLookup: &api.MsgLookup{
+				Receipt: types.MessageReceipt{
+					ExitCode: exitcode.Ok,
+					Return:   newValueReturn,
+				},
+			},
+			checkTsDeals: map[abi.DealID]*api.MarketDeal{
+				startDealID: unfinishedDeal,
+			},
+			matchStates: []matchState{
+				{
+					msg: makeMessage(t, provider, miner.Methods.PreCommitSector, &miner.SectorPreCommitInfo{
+						SectorNumber: sectorNumber,
+						SealedCID:    sealedCid,
+						DealIDs:      []abi.DealID{startDealID},
+					}),
+					deals: map[abi.DealID]*api.MarketDeal{
+						startDealID: unfinishedDeal,
+					},
+				},
+				{
+					msg: makeMessage(t, provider, miner.Methods.ProveCommitSector, &miner.ProveCommitSectorParams{
+						SectorNumber: sectorNumber,
+					}),
+					deals: map[abi.DealID]*api.MarketDeal{
+						newDealID: successDeal,
+					},
+				},
+			},
+			expectedCBCallCount: 1,
+		},
+		"prove commit but no sector recorded": {
+			checkTsDeals: map[abi.DealID]*api.MarketDeal{
+				startDealID: unfinishedDeal,
+			},
+			matchStates: []matchState{
+				{
+					msg: makeMessage(t, provider, miner.Methods.ProveCommitSector, &miner.ProveCommitSectorParams{
+						SectorNumber: sectorNumber,
+					}),
+					deals: map[abi.DealID]*api.MarketDeal{
+						startDealID: successDeal,
+					},
+				},
+			},
+			expectedCBCallCount: 0,
+		},
+		"error on deal in check": {
+			checkTsDeals:        map[abi.DealID]*api.MarketDeal{},
+			searchMessageErr:    errors.New("something went wrong"),
+			expectedCBCallCount: 0,
+			expectedError:       errors.New("failed to set up called handler: failed to look up deal on chain: something went wrong"),
+		},
+		"sector start epoch > 0 in check": {
+			checkTsDeals: map[abi.DealID]*api.MarketDeal{
+				startDealID: successDeal,
+			},
+			expectedCBCallCount: 1,
+		},
+		"error on deal in pre-commit": {
+			searchMessageErr: errors.New("something went wrong"),
+			checkTsDeals: map[abi.DealID]*api.MarketDeal{
+				startDealID: unfinishedDeal,
+			},
+			matchStates: []matchState{
+				{
+					msg: makeMessage(t, provider, miner.Methods.PreCommitSector, &miner.SectorPreCommitInfo{
+						SectorNumber: sectorNumber,
+						SealedCID:    sealedCid,
+						DealIDs:      []abi.DealID{startDealID},
+					}),
+					deals: map[abi.DealID]*api.MarketDeal{},
+				},
+			},
+			expectedCBCallCount: 1,
+			expectedCBError:     errors.New("handling applied event: something went wrong"),
+			expectedError:       errors.New("failed to set up called handler: something went wrong"),
+		},
+	}
+	runTestCase := func(testCase string, data struct {
+		searchMessageLookup *api.MsgLookup
+		searchMessageErr    error
+		checkTsDeals        map[abi.DealID]*api.MarketDeal
+		matchStates         []matchState
+		expectedCBCallCount uint64
+		expectedCBError     error
+		expectedError       error
+	}) {
+		t.Run(testCase, func(t *testing.T) {
+			//	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+			//	defer cancel()
+			api := &mockGetCurrentDealInfoAPI{
+				SearchMessageLookup: data.searchMessageLookup,
+				SearchMessageErr:    data.searchMessageErr,
+				MarketDeals:         make(map[marketDealKey]*api.MarketDeal),
+			}
+			checkTs, err := test.MockTipset(provider, rand.Uint64())
+			require.NoError(t, err)
+			for dealID, deal := range data.checkTsDeals {
+				api.MarketDeals[marketDealKey{dealID, checkTs.Key()}] = deal
+			}
+			matchMessages := make([]matchMessage, len(data.matchStates))
+			for i, ms := range data.matchStates {
+				matchTs, err := test.MockTipset(provider, rand.Uint64())
+				require.NoError(t, err)
+				for dealID, deal := range ms.deals {
+					api.MarketDeals[marketDealKey{dealID, matchTs.Key()}] = deal
+				}
+				matchMessages[i] = matchMessage{
+					curH:       5,
+					msg:        ms.msg,
+					msgReceipt: nil,
+					ts:         matchTs,
+				}
+			}
+			eventsAPI := &fakeEvents{
+				Ctx:           ctx,
+				CheckTs:       checkTs,
+				MatchMessages: matchMessages,
+			}
+			cbCallCount := uint64(0)
+			var cbError error
+			cb := func(err error) {
+				cbCallCount++
+				cbError = err
+			}
+			err = OnDealSectorCommitted(ctx, api, eventsAPI, provider, startDealID, proposal, &publishCid, cb)
+			if data.expectedError == nil {
+				require.NoError(t, err)
+			} else {
+				require.EqualError(t, err, data.expectedError.Error())
+			}
+			require.Equal(t, data.expectedCBCallCount, cbCallCount)
+			if data.expectedCBError == nil {
+				require.NoError(t, cbError)
+			} else {
+				require.EqualError(t, cbError, data.expectedCBError.Error())
+			}
+		})
+	}
+	for testCase, data := range testCases {
+		runTestCase(testCase, data)
+	}
+}
+
+type matchState struct {
+	msg   *types.Message
+	deals map[abi.DealID]*api.MarketDeal
+}
+
+type matchMessage struct {
+	curH       abi.ChainEpoch
+	msg        *types.Message
+	msgReceipt *types.MessageReceipt
+	ts         *types.TipSet
+	doesRevert bool
+}
+type fakeEvents struct {
+	Ctx           context.Context
+	CheckTs       *types.TipSet
+	MatchMessages []matchMessage
+}
+
+func (fe *fakeEvents) Called(check events.CheckFunc, msgHnd events.MsgHandler, rev events.RevertHandler, confidence int, timeout abi.ChainEpoch, mf events.MsgMatchFunc) error {
+	_, more, err := check(fe.CheckTs)
+	if err != nil {
+		return err
+	}
+	if !more {
+		return nil
+	}
+	for _, matchMessage := range fe.MatchMessages {
+		matched, err := mf(matchMessage.msg)
+		if err != nil {
+			return err
+		}
+		if matched {
+			more, err := msgHnd(matchMessage.msg, matchMessage.msgReceipt, matchMessage.ts, matchMessage.curH)
+			if err != nil {
+				return err
+			}
+			if matchMessage.doesRevert {
+				err := rev(fe.Ctx, matchMessage.ts)
+				if err != nil {
+					return err
+				}
+			}
+			if !more {
+				return nil
+			}
+		}
+	}
+	return nil
+}
+
+func makeMessage(t *testing.T, to address.Address, method abi.MethodNum, params cbor.Marshaler) *types.Message {
+	buf := new(bytes.Buffer)
+	err := params.MarshalCBOR(buf)
+	require.NoError(t, err)
+	return &types.Message{
+		To:     to,
+		Method: method,
+		Params: buf.Bytes(),
+	}
+}
+
+var seq int
+
+func generateCids(n int) []cid.Cid {
+	cids := make([]cid.Cid, 0, n)
+	for i := 0; i < n; i++ {
+		c := blocks.NewBlock([]byte(fmt.Sprint(seq))).Cid()
+		seq++
+		cids = append(cids, c)
+	}
+	return cids
+}

markets/storageadapter/provider.go

@@ -3,7 +3,6 @@ package storageadapter
 // this file implements storagemarket.StorageProviderNode
 
 import (
-	"bytes"
 	"context"
 	"io"
 	"time"
@@ -25,7 +24,6 @@ import (
 	"github.com/filecoin-project/lotus/build"
 	"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/events"
 	"github.com/filecoin-project/lotus/chain/events/state"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -255,107 +253,8 @@ func (n *ProviderNodeAdapter) DealProviderCollateralBounds(ctx context.Context,
 	return bounds.Min, bounds.Max, nil
 }
 
-func (n *ProviderNodeAdapter) OnDealSectorCommitted(ctx context.Context, provider address.Address, dealID abi.DealID, cb storagemarket.DealSectorCommittedCallback) error {
+func (n *ProviderNodeAdapter) OnDealSectorCommitted(ctx context.Context, provider address.Address, dealID abi.DealID, proposal market2.DealProposal, publishCid *cid.Cid, cb storagemarket.DealSectorCommittedCallback) error {
-	checkFunc := func(ts *types.TipSet) (done bool, more bool, err error) {
+	return OnDealSectorCommitted(ctx, n, n.ev, provider, dealID, market.DealProposal(proposal), publishCid, cb)
-		sd, err := n.StateMarketStorageDeal(ctx, dealID, ts.Key())
-
-		if err != nil {
-			// TODO: This may be fine for some errors
-			return false, false, xerrors.Errorf("failed to look up deal on chain: %w", err)
-		}
-
-		if sd.State.SectorStartEpoch > 0 {
-			cb(nil)
-			return true, false, nil
-		}
-
-		return false, true, nil
-	}
-
-	called := func(msg *types.Message, rec *types.MessageReceipt, ts *types.TipSet, curH abi.ChainEpoch) (more bool, err error) {
-		defer func() {
-			if err != nil {
-				cb(xerrors.Errorf("handling applied event: %w", err))
-			}
-		}()
-
-		if msg == nil {
-			log.Error("timed out waiting for deal activation... what now?")
-			return false, nil
-		}
-
-		sd, err := n.StateMarketStorageDeal(ctx, dealID, ts.Key())
-		if err != nil {
-			return false, xerrors.Errorf("failed to look up deal on chain: %w", err)
-		}
-
-		if sd.State.SectorStartEpoch < 1 {
-			return false, xerrors.Errorf("deal wasn't active: deal=%d, parentState=%s, h=%d", dealID, ts.ParentState(), ts.Height())
-		}
-
-		log.Infof("Storage deal %d activated at epoch %d", dealID, sd.State.SectorStartEpoch)
-
-		cb(nil)
-
-		return false, nil
-	}
-
-	revert := func(ctx context.Context, ts *types.TipSet) error {
-		log.Warn("deal activation reverted; TODO: actually handle this!")
-		// TODO: Just go back to DealSealing?
-		return nil
-	}
-
-	var sectorNumber abi.SectorNumber
-	var sectorFound bool
-
-	matchEvent := func(msg *types.Message) (matched bool, err error) {
-		if msg.To != provider {
-			return false, nil
-		}
-
-		switch msg.Method {
-		case miner.Methods.PreCommitSector:
-			var params miner.SectorPreCommitInfo
-			if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
-				return false, xerrors.Errorf("unmarshal pre commit: %w", err)
-			}
-
-			for _, did := range params.DealIDs {
-				if did == dealID {
-					sectorNumber = params.SectorNumber
-					sectorFound = true
-					return false, nil
-				}
-			}
-
-			return false, nil
-		case miner.Methods.ProveCommitSector:
-			var params miner.ProveCommitSectorParams
-			if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
-				return false, xerrors.Errorf("failed to unmarshal prove commit sector params: %w", err)
-			}
-
-			if !sectorFound {
-				return false, nil
-			}
-
-			if params.SectorNumber != sectorNumber {
-				return false, nil
-			}
-
-			return true, nil
-		default:
-			return false, nil
-		}
-
-	}
-
-	if err := n.ev.Called(checkFunc, called, revert, int(build.MessageConfidence+1), events.NoTimeout, matchEvent); err != nil {
-		return xerrors.Errorf("failed to set up called handler: %w", err)
-	}
-
-	return nil
 }
 
 func (n *ProviderNodeAdapter) GetChainHead(ctx context.Context) (shared.TipSetToken, abi.ChainEpoch, error) {