refactor(markets): refactor OnDealSectorCommitted

extract OnDealSectorCommitted from adapters and test

markets/storageadapter/client.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
-	"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/market"
@@ -214,115 +213,7 @@ func (c *ClientNodeAdapter) DealProviderCollateralBounds(ctx context.Context, si
 }
 
 func (c *ClientNodeAdapter) OnDealSectorCommitted(ctx context.Context, provider address.Address, dealID abi.DealID, publishCid *cid.Cid, cb storagemarket.DealSectorCommittedCallback) error {
-	checkFunc := func(ts *types.TipSet) (done bool, more bool, err error) {
-		newDealID, sd, err := GetCurrentDealInfo(ctx, ts, c, dealID, 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:
-			dealID, _, err = GetCurrentDealInfo(ctx, ts, c, dealID, publishCid)
-			if err != nil {
-				return false, err
-			}
-
-			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 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 := 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
-		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 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
+	return OnDealSectorCommitted(c, c.ev, ctx, provider, dealID, publishCid, cb)
 }
 
 func (c *ClientNodeAdapter) OnDealExpiredOrSlashed(ctx context.Context, dealID abi.DealID, onDealExpired storagemarket.DealExpiredCallback, onDealSlashed storagemarket.DealSlashedCallback) error {

markets/storageadapter/getcurrentdealinfo_test.go

@@ -137,10 +137,14 @@ func TestGetCurrentDealInfo(t *testing.T) {
 			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:         data.marketDeals,
+				MarketDeals:         marketDeals,
 			}
 
 			dealID, marketDeal, err := GetCurrentDealInfo(ctx, ts, api, startDealID, data.publishCid)
@@ -155,15 +159,20 @@ func TestGetCurrentDealInfo(t *testing.T) {
 	}
 }
 
+type marketDealKey struct {
+	abi.DealID
+	types.TipSetKey
+}
+
 type mockGetCurrentDealInfoAPI struct {
 	SearchMessageLookup *api.MsgLookup
 	SearchMessageErr    error
 
-	MarketDeals map[abi.DealID]*api.MarketDeal
+	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[dealID]
+	deal, ok := mapi.MarketDeals[marketDealKey{dealID, ts}]
 	if !ok {
 		return nil, notFoundError
 	}

markets/storageadapter/ondealsectorcommitted.go

@@ -0,0 +1,133 @@
+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/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(api getCurrentDealInfoAPI, eventsApi sectorCommittedEventsAPI, ctx context.Context, provider address.Address, dealID abi.DealID, 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, 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:
+			dealID, _, err = GetCurrentDealInfo(ctx, ts, api, dealID, publishCid)
+			if err != nil {
+				return false, err
+			}
+
+			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 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, 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:
+			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 := 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,353 @@
+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]
+	startDealID := abi.DealID(rand.Uint64())
+	newDealID := abi.DealID(rand.Uint64())
+	newValueReturn := makePublishDealsReturnBytes(t, []abi.DealID{newDealID})
+	sectorNumber := abi.SectorNumber(rand.Uint64())
+	unfinishedDeal := &api.MarketDeal{
+		State: market.DealState{
+			SectorStartEpoch: -1,
+			LastUpdatedEpoch: 2,
+		},
+	}
+	successDeal := &api.MarketDeal{
+		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"),
+		},
+	}
+	for testCase, data := range testCases {
+		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(api, eventsAPI, ctx, provider, startDealID, &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())
+			}
+		})
+	}
+}
+
+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(string(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"
@@ -256,116 +254,7 @@ func (n *ProviderNodeAdapter) DealProviderCollateralBounds(ctx context.Context,
 }
 
 func (n *ProviderNodeAdapter) OnDealSectorCommitted(ctx context.Context, provider address.Address, dealID abi.DealID, publishCid *cid.Cid, cb storagemarket.DealSectorCommittedCallback) error {
-	checkFunc := func(ts *types.TipSet) (done bool, more bool, err error) {
-		newDealID, sd, err := GetCurrentDealInfo(ctx, ts, n, dealID, 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:
-			dealID, _, err = GetCurrentDealInfo(ctx, ts, n, dealID, publishCid)
-			if err != nil {
-				return false, err
-			}
-
-			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 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 := 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
-		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 true, 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
+	return OnDealSectorCommitted(n, n.ev, ctx, provider, dealID, publishCid, cb)
 }
 
 func (n *ProviderNodeAdapter) GetChainHead(ctx context.Context) (shared.TipSetToken, abi.ChainEpoch, error) {