test: predicates test

chain/events/state/predicates.go

@@ -17,8 +17,7 @@ import (
 type UserData interface{}
 
 type ChainApi interface {
-	ChainHasObj(context.Context, cid.Cid) (bool, error)
+	apibstore.ChainIO
-	ChainReadObj(context.Context, cid.Cid) ([]byte, error)
 	StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error)
 }
 
@@ -54,7 +53,7 @@ func (sp *StatePredicates) OnActorStateChanged(addr address.Address, diffStateFu
 
 type DiffStorageMarketStateFunc func(ctx context.Context, oldState *market.State, newState *market.State) (changed bool, user UserData, err error)
 
-func (sp *StatePredicates) OnStorageMarketActorChanged(addr address.Address, diffStorageMarketState DiffStorageMarketStateFunc) DiffFunc {
+func (sp *StatePredicates) OnStorageMarketActorChanged(diffStorageMarketState DiffStorageMarketStateFunc) DiffFunc {
 	return sp.OnActorStateChanged(builtin.StorageMarketActorAddr, func(ctx context.Context, oldActorStateHead, newActorStateHead cid.Cid) (changed bool, user UserData, err error) {
 		var oldState market.State
 		if err := sp.cst.Get(ctx, oldActorStateHead, &oldState); err != nil {
@@ -89,9 +88,16 @@ func (sp *StatePredicates) OnDealStateChanged(diffDealStates DiffDealStatesFunc)
 	}
 }
 
+type ChangedDeals map[abi.DealID]DealStateChange
+
+type DealStateChange struct {
+	From market.DealState
+	To market.DealState
+}
+
 func (sp *StatePredicates) DealStateChangedForIDs(dealIds []abi.DealID) DiffDealStatesFunc {
 	return func(ctx context.Context, oldDealStateRoot *amt.Root, newDealStateRoot *amt.Root) (changed bool, user UserData, err error) {
-		var changedDeals []abi.DealID
+		changedDeals := make(ChangedDeals)
 		for _, dealId := range dealIds {
 			var oldDeal, newDeal market.DealState
 			err := oldDealStateRoot.Get(ctx, uint64(dealId), &oldDeal)
@@ -103,11 +109,11 @@ func (sp *StatePredicates) DealStateChangedForIDs(dealIds []abi.DealID) DiffDeal
 				return false, nil, err
 			}
 			if oldDeal != newDeal {
-				changedDeals = append(changedDeals, dealId)
+				changedDeals[dealId] = DealStateChange{oldDeal, newDeal}
 			}
 		}
 		if len(changedDeals) > 0 {
-			return true, changed, nil
+			return true, changedDeals, nil
 		}
 		return false, nil, nil
 	}

chain/events/state/predicates_test.go

@@ -0,0 +1,176 @@
+package state
+
+import (
+	"context"
+	"github.com/filecoin-project/specs-actors/actors/crypto"
+	"github.com/ipfs/go-hamt-ipld"
+	"testing"
+
+	"github.com/filecoin-project/go-amt-ipld/v2"
+	"github.com/filecoin-project/specs-actors/actors/builtin/market"
+	ds "github.com/ipfs/go-datastore"
+	ds_sync "github.com/ipfs/go-datastore/sync"
+	bstore "github.com/ipfs/go-ipfs-blockstore"
+	cbornode "github.com/ipfs/go-ipld-cbor"
+	"golang.org/x/xerrors"
+
+	"github.com/ipfs/go-cid"
+	"github.com/stretchr/testify/require"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/specs-actors/actors/abi"
+)
+
+var dummyCid cid.Cid
+
+func init() {
+	dummyCid, _ = cid.Parse("bafkqaaa")
+}
+
+type mockApi struct {
+	ts map[types.TipSetKey]*types.Actor
+	bs bstore.Blockstore
+}
+
+func newMockApi(bs bstore.Blockstore) *mockApi {
+	return &mockApi{
+		bs: bs,
+		ts: make(map[types.TipSetKey]*types.Actor),
+	}
+}
+
+func (m mockApi) ChainHasObj(ctx context.Context, c cid.Cid) (bool, error) {
+	return m.bs.Has(c)
+}
+
+func (m mockApi) ChainReadObj(ctx context.Context, c cid.Cid) ([]byte, error) {
+	blk, err := m.bs.Get(c)
+	if err != nil {
+		return nil, xerrors.Errorf("blockstore get: %w", err)
+	}
+
+	return blk.RawData(), nil
+}
+
+func (m mockApi) StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error) {
+	return m.ts[tsk], nil
+}
+
+func (m mockApi) setActor(tsk types.TipSetKey, act *types.Actor) {
+	m.ts[tsk] = act
+}
+
+func TestPredicates(t *testing.T) {
+	ctx := context.Background()
+
+	bs := bstore.NewBlockstore(ds_sync.MutexWrap(ds.NewMapDatastore()))
+	store := cbornode.NewCborStore(bs)
+
+	oldDeals := map[abi.DealID]*market.DealState{
+		abi.DealID(1): {
+			SectorStartEpoch: 1,
+			LastUpdatedEpoch: 2,
+			SlashEpoch:       0,
+		},
+		abi.DealID(2): {
+			SectorStartEpoch: 4,
+			LastUpdatedEpoch: 5,
+			SlashEpoch:       0,
+		},
+	}
+	oldStateC := createMarketState(t, store, oldDeals, ctx)
+
+	newDeals := map[abi.DealID]*market.DealState{
+		abi.DealID(1): {
+			SectorStartEpoch: 1,
+			LastUpdatedEpoch: 3,
+			SlashEpoch:       0,
+		},
+		abi.DealID(2): {
+			SectorStartEpoch: 4,
+			LastUpdatedEpoch: 6,
+			SlashEpoch:       6,
+		},
+	}
+	newStateC := createMarketState(t, store, newDeals, ctx)
+
+	miner, err := address.NewFromString("t00")
+	require.NoError(t, err)
+	oldState, err := mockTipset(miner, 1)
+	require.NoError(t, err)
+	newState, err := mockTipset(miner, 2)
+	require.NoError(t, err)
+
+	api := newMockApi(bs)
+	api.setActor(oldState.Key(), &types.Actor{Head: oldStateC})
+	api.setActor(newState.Key(), &types.Actor{Head: newStateC})
+
+	preds := NewStatePredicates(api)
+
+	dealIds := []abi.DealID{abi.DealID(1), abi.DealID(2)}
+	diffFn := preds.OnStorageMarketActorChanged(preds.OnDealStateChanged(preds.DealStateChangedForIDs(dealIds)))
+
+	// Diff a state against itself: expect no change
+	changed, _, err := diffFn(ctx, oldState, oldState)
+	require.NoError(t, err)
+	require.False(t, changed)
+
+	// Diff old state against new state
+	changed, val, err := diffFn(ctx, oldState, newState)
+	require.NoError(t, err)
+	require.True(t, changed)
+
+	changedDeals, ok := val.(ChangedDeals)
+	require.True(t, ok)
+	require.Len(t, changedDeals, 2)
+	require.Contains(t, changedDeals, abi.DealID(1))
+	require.Contains(t, changedDeals, abi.DealID(2))
+	deal1 := changedDeals[abi.DealID(1)]
+	if deal1.From.LastUpdatedEpoch != 2 || deal1.To.LastUpdatedEpoch != 3 {
+		t.Fatal("Unexpected change to LastUpdatedEpoch")
+	}
+	deal2 := changedDeals[abi.DealID(2)]
+	if deal2.From.SlashEpoch != 0 || deal2.To.SlashEpoch != 6 {
+		t.Fatal("Unexpected change to LastUpdatedEpoch")
+	}
+}
+
+func mockTipset(miner address.Address, timestamp uint64) (*types.TipSet, error) {
+	return types.NewTipSet([]*types.BlockHeader{{
+		Miner:                 miner,
+		Height:                5,
+		ParentStateRoot:       dummyCid,
+		Messages:              dummyCid,
+		ParentMessageReceipts: dummyCid,
+		BlockSig:              &crypto.Signature{Type: crypto.SigTypeBLS},
+		BLSAggregate:          &crypto.Signature{Type: crypto.SigTypeBLS},
+		Timestamp:             timestamp,
+	}})
+}
+
+func createMarketState(t *testing.T, store *cbornode.BasicIpldStore, deals map[abi.DealID]*market.DealState, ctx context.Context) cid.Cid {
+	rootCid := createAMT(t, store, deals, ctx)
+
+	emptyArrayCid, err := amt.NewAMT(store).Flush(context.TODO())
+	require.NoError(t, err)
+	emptyMap, err := store.Put(context.TODO(), hamt.NewNode(store, hamt.UseTreeBitWidth(5)))
+	require.NoError(t, err)
+	state := market.ConstructState(emptyArrayCid, emptyMap, emptyMap)
+	state.States = rootCid
+
+	stateC, err := store.Put(ctx, state)
+	require.NoError(t, err)
+	return stateC
+}
+
+func createAMT(t *testing.T, store *cbornode.BasicIpldStore, deals map[abi.DealID]*market.DealState, ctx context.Context) cid.Cid {
+	root := amt.NewAMT(store)
+	for dealId, dealState := range deals {
+		err := root.Set(ctx, uint64(dealId), dealState)
+		require.NoError(t, err)
+	}
+	rootCid, err := root.Flush(ctx)
+	require.NoError(t, err)
+	return rootCid
+}