Merge pull request #3013 from filecoin-project/frrist/fix-id-map-table

fix: id_address_map conflicts

chain/events/state/diff_adt.go

@@ -1,6 +1,7 @@
 package state
 
 import (
+	"bytes"
 	"github.com/filecoin-project/specs-actors/actors/util/adt"
 	typegen "github.com/whyrusleeping/cbor-gen"
 )
@@ -39,8 +40,11 @@ func DiffAdtArray(preArr, curArr *adt.Array, out AdtArrayDiff) error {
 			return nil
 		}
 
-		if err := out.Modify(uint64(i), prevVal, curVal); err != nil {
+		// no modification
-			return err
+		if !bytes.Equal(prevVal.Raw,curVal.Raw)  {
+			if err := out.Modify(uint64(i), prevVal, curVal); err != nil {
+				return err
+			}
 		}
 
 		return curArr.Delete(uint64(i))
@@ -90,10 +94,14 @@ func DiffAdtMap(preMap, curMap *adt.Map, out AdtMapDiff) error {
 			return nil
 		}
 
-		if err := out.Modify(key, prevVal, curVal); err != nil {
+		// no modification
-			return err
+		if !bytes.Equal(prevVal.Raw,curVal.Raw) {
+			if err := out.Modify(key, prevVal, curVal); err != nil {
+				return err
+			}
 		}
 
+
 		return curMap.Delete(k)
 	}); err != nil {
 		return err

chain/events/state/diff_adt_test.go

@@ -40,7 +40,7 @@ func TestDiffAdtArray(t *testing.T) {
 	require.NoError(t, arrB.Set(5, runtime.CBORBytes{8})) // add
 	require.NoError(t, arrB.Set(6, runtime.CBORBytes{9})) // add
 
-	changes := new(TestAdtDiff)
+	changes := new(TestDiffArray)
 
 	assert.NoError(t, DiffAdtArray(arrA, arrB, changes))
 	assert.NotNil(t, changes)
@@ -71,38 +71,188 @@ func TestDiffAdtArray(t *testing.T) {
 	assert.EqualValues(t, []byte{1}, changes.Removed[1].val)
 }
 
-type adtDiffResult struct {
+
-	key uint64
+func TestDiffAdtMap(t *testing.T) {
-	val runtime.CBORBytes
+	ctxstoreA := newContextStore()
+	ctxstoreB := newContextStore()
+
+	mapA := adt.MakeEmptyMap(ctxstoreA)
+	mapB := adt.MakeEmptyMap(ctxstoreB)
+
+	require.NoError(t, mapA.Put(adt.UIntKey(0), runtime.CBORBytes([]byte{0}))) // delete
+
+	require.NoError(t, mapA.Put(adt.UIntKey(1), runtime.CBORBytes([]byte{0}))) // modify
+	require.NoError(t, mapB.Put(adt.UIntKey(1), runtime.CBORBytes([]byte{1})))
+
+	require.NoError(t, mapA.Put(adt.UIntKey(2), runtime.CBORBytes([]byte{1}))) // delete
+
+	require.NoError(t, mapA.Put(adt.UIntKey(3), runtime.CBORBytes([]byte{0}))) // noop
+	require.NoError(t, mapB.Put(adt.UIntKey(3), runtime.CBORBytes([]byte{0})))
+
+	require.NoError(t, mapA.Put(adt.UIntKey(4), runtime.CBORBytes([]byte{0}))) // modify
+	require.NoError(t, mapB.Put(adt.UIntKey(4), runtime.CBORBytes([]byte{6})))
+
+	require.NoError(t, mapB.Put(adt.UIntKey(5), runtime.CBORBytes{8})) // add
+	require.NoError(t, mapB.Put(adt.UIntKey(6), runtime.CBORBytes{9})) // add
+
+	changes := new(TestDiffMap)
+
+	assert.NoError(t, DiffAdtMap(mapA, mapB, changes))
+	assert.NotNil(t, changes)
+
+	assert.Equal(t, 2, len(changes.Added))
+	// keys 5 and 6 were added
+	assert.EqualValues(t, uint64(6), changes.Added[0].key)
+	assert.EqualValues(t, []byte{9}, changes.Added[0].val)
+	assert.EqualValues(t, uint64(5), changes.Added[1].key)
+	assert.EqualValues(t, []byte{8}, changes.Added[1].val)
+
+	assert.Equal(t, 2, len(changes.Modified))
+	// keys 1 and 4 were modified
+	assert.EqualValues(t, uint64(1), changes.Modified[0].From.key)
+	assert.EqualValues(t, []byte{0}, changes.Modified[0].From.val)
+	assert.EqualValues(t, uint64(1), changes.Modified[0].To.key)
+	assert.EqualValues(t, []byte{1}, changes.Modified[0].To.val)
+	assert.EqualValues(t, uint64(4), changes.Modified[1].From.key)
+	assert.EqualValues(t, []byte{0}, changes.Modified[1].From.val)
+	assert.EqualValues(t, uint64(4), changes.Modified[1].To.key)
+	assert.EqualValues(t, []byte{6}, changes.Modified[1].To.val)
+
+	assert.Equal(t, 2, len(changes.Removed))
+	// keys 0 and 2 were deleted
+	assert.EqualValues(t, uint64(0), changes.Removed[0].key)
+	assert.EqualValues(t, []byte{0}, changes.Removed[0].val)
+	assert.EqualValues(t, uint64(2), changes.Removed[1].key)
+	assert.EqualValues(t, []byte{1}, changes.Removed[1].val)
+
 }
 
-type TestAdtDiff struct {
+type TestDiffMap struct {
-	Added    []adtDiffResult
+	Added []adtMapDiffResult
-	Modified []TestAdtDiffModified
+	Modified []TestAdtMapDiffModified
-	Removed  []adtDiffResult
+	Removed []adtMapDiffResult
 }
 
-var _ AdtArrayDiff = &TestAdtDiff{}
+var _ AdtMapDiff = &TestDiffMap{}
 
-type TestAdtDiffModified struct {
+
-	From adtDiffResult
+func (t *TestDiffMap) AsKey(key string) (adt.Keyer, error) {
-	To   adtDiffResult
+	k, err := adt.ParseUIntKey(key)
+	if err != nil {
+		return nil, err
+	}
+	return adt.UIntKey(k), nil
 }
 
-func (t *TestAdtDiff) Add(key uint64, val *typegen.Deferred) error {
+func (t *TestDiffMap) Add(key string, val *typegen.Deferred) error {
 	v := new(runtime.CBORBytes)
 	err := v.UnmarshalCBOR(bytes.NewReader(val.Raw))
 	if err != nil {
 		return err
 	}
-	t.Added = append(t.Added, adtDiffResult{
+	k, err := adt.ParseUIntKey(key)
+	if err != nil {
+		return err
+	}
+	t.Added = append(t.Added, adtMapDiffResult{
+		key: k,
+		val: *v,
+	})
+	return nil
+}
+
+func (t *TestDiffMap) Modify(key string, from, to *typegen.Deferred) error {
+	vFrom := new(runtime.CBORBytes)
+	err := vFrom.UnmarshalCBOR(bytes.NewReader(from.Raw))
+	if err != nil {
+		return err
+	}
+
+	vTo := new(runtime.CBORBytes)
+	err = vTo.UnmarshalCBOR(bytes.NewReader(to.Raw))
+	if err != nil {
+		return err
+	}
+
+	k, err := adt.ParseUIntKey(key)
+	if err != nil {
+		return err
+	}
+
+	if !bytes.Equal(*vFrom, *vTo) {
+		t.Modified = append(t.Modified, TestAdtMapDiffModified{
+			From: adtMapDiffResult{
+				key: k,
+				val: *vFrom,
+			},
+			To: adtMapDiffResult{
+				key: k,
+				val: *vTo,
+			},
+		})
+	}
+	return nil
+}
+
+func (t *TestDiffMap) Remove(key string, val *typegen.Deferred) error {
+	v := new(runtime.CBORBytes)
+	err := v.UnmarshalCBOR(bytes.NewReader(val.Raw))
+	if err != nil {
+		return err
+	}
+	k, err := adt.ParseUIntKey(key)
+	if err != nil {
+		return err
+	}
+	t.Removed = append(t.Removed, adtMapDiffResult{
+		key: k,
+		val: *v,
+	})
+	return nil
+}
+
+type adtMapDiffResult struct {
+	key uint64
+	val runtime.CBORBytes
+}
+
+type TestAdtMapDiffModified struct {
+	From adtMapDiffResult
+	To adtMapDiffResult
+}
+
+type adtArrayDiffResult struct {
+	key uint64
+	val runtime.CBORBytes
+}
+
+type TestDiffArray struct {
+	Added    []adtArrayDiffResult
+	Modified []TestAdtArrayDiffModified
+	Removed  []adtArrayDiffResult
+}
+
+var _ AdtArrayDiff = &TestDiffArray{}
+
+type TestAdtArrayDiffModified struct {
+	From adtArrayDiffResult
+	To   adtArrayDiffResult
+}
+
+func (t *TestDiffArray) Add(key uint64, val *typegen.Deferred) error {
+	v := new(runtime.CBORBytes)
+	err := v.UnmarshalCBOR(bytes.NewReader(val.Raw))
+	if err != nil {
+		return err
+	}
+	t.Added = append(t.Added, adtArrayDiffResult{
 		key: key,
 		val: *v,
 	})
 	return nil
 }
 
-func (t *TestAdtDiff) Modify(key uint64, from, to *typegen.Deferred) error {
+func (t *TestDiffArray) Modify(key uint64, from, to *typegen.Deferred) error {
 	vFrom := new(runtime.CBORBytes)
 	err := vFrom.UnmarshalCBOR(bytes.NewReader(from.Raw))
 	if err != nil {
@@ -116,12 +266,12 @@ func (t *TestAdtDiff) Modify(key uint64, from, to *typegen.Deferred) error {
 	}
 
 	if !bytes.Equal(*vFrom, *vTo) {
-		t.Modified = append(t.Modified, TestAdtDiffModified{
+		t.Modified = append(t.Modified, TestAdtArrayDiffModified{
-			From: adtDiffResult{
+			From: adtArrayDiffResult{
 				key: key,
 				val: *vFrom,
 			},
-			To: adtDiffResult{
+			To: adtArrayDiffResult{
 				key: key,
 				val: *vTo,
 			},
@@ -130,13 +280,13 @@ func (t *TestAdtDiff) Modify(key uint64, from, to *typegen.Deferred) error {
 	return nil
 }
 
-func (t *TestAdtDiff) Remove(key uint64, val *typegen.Deferred) error {
+func (t *TestDiffArray) Remove(key uint64, val *typegen.Deferred) error {
 	v := new(runtime.CBORBytes)
 	err := v.UnmarshalCBOR(bytes.NewReader(val.Raw))
 	if err != nil {
 		return err
 	}
-	t.Removed = append(t.Removed, adtDiffResult{
+	t.Removed = append(t.Removed, adtArrayDiffResult{
 		key: key,
 		val: *v,
 	})

chain/events/state/predicates.go

@@ -3,19 +3,18 @@ package state
 import (
 	"bytes"
 	"context"
-
 	"github.com/filecoin-project/go-address"
-	"github.com/ipfs/go-cid"
-	cbor "github.com/ipfs/go-ipld-cbor"
-	typegen "github.com/whyrusleeping/cbor-gen"
-
 	"github.com/filecoin-project/specs-actors/actors/abi"
 	"github.com/filecoin-project/specs-actors/actors/abi/big"
 	"github.com/filecoin-project/specs-actors/actors/builtin"
+	init_ "github.com/filecoin-project/specs-actors/actors/builtin/init"
 	"github.com/filecoin-project/specs-actors/actors/builtin/market"
 	"github.com/filecoin-project/specs-actors/actors/builtin/miner"
 	"github.com/filecoin-project/specs-actors/actors/builtin/paych"
 	"github.com/filecoin-project/specs-actors/actors/util/adt"
+	"github.com/ipfs/go-cid"
+	cbor "github.com/ipfs/go-ipld-cbor"
+	typegen "github.com/whyrusleeping/cbor-gen"
 
 	"github.com/filecoin-project/lotus/api/apibstore"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -407,6 +406,21 @@ func (sp *StatePredicates) AvailableBalanceChangedForAddresses(getAddrs func() [
 
 type DiffMinerActorStateFunc func(ctx context.Context, oldState *miner.State, newState *miner.State) (changed bool, user UserData, err error)
 
+func (sp *StatePredicates) OnInitActorChange(diffInitActorState DiffInitActorStateFunc) DiffTipSetKeyFunc {
+	return sp.OnActorStateChanged(builtin.InitActorAddr, func(ctx context.Context, oldActorStateHead, newActorStateHead cid.Cid) (changed bool, user UserData, err error) {
+		var oldState init_.State
+		if err := sp.cst.Get(ctx, oldActorStateHead, &oldState); err != nil {
+			return false, nil, err
+		}
+		var newState init_.State
+		if err := sp.cst.Get(ctx, newActorStateHead, &newState); err != nil {
+			return false, nil, err
+		}
+		return diffInitActorState(ctx, &oldState, &newState)
+	})
+
+}
+
 func (sp *StatePredicates) OnMinerActorChange(minerAddr address.Address, diffMinerActorState DiffMinerActorStateFunc) DiffTipSetKeyFunc {
 	return sp.OnActorStateChanged(minerAddr, func(ctx context.Context, oldActorStateHead, newActorStateHead cid.Cid) (changed bool, user UserData, err error) {
 		var oldState miner.State
@@ -628,3 +642,148 @@ func (sp *StatePredicates) OnToSendAmountChanges() DiffPaymentChannelStateFunc {
 		}, nil
 	}
 }
+
+type AddressPair struct {
+	ID address.Address
+	PK address.Address
+}
+
+type InitActorAddressChanges struct {
+	Added []AddressPair
+	Modified []AddressChange
+	Removed []AddressPair
+}
+
+type AddressChange struct {
+	From AddressPair
+	To AddressPair
+}
+
+type DiffInitActorStateFunc func(ctx context.Context, oldState *init_.State, newState *init_.State) (changed bool, user UserData, err error)
+
+
+
+func (i *InitActorAddressChanges) AsKey(key string) (adt.Keyer, error) {
+	addr , err := address.NewFromBytes([]byte(key))
+	if err != nil {
+		return nil, err
+	}
+	return adt.AddrKey(addr), nil
+}
+
+func (i *InitActorAddressChanges) Add(key string, val *typegen.Deferred) error {
+	pkAddr, err := address.NewFromBytes([]byte(key))
+	if err != nil {
+		return err
+	}
+	id := new(typegen.CborInt)
+	if err := id.UnmarshalCBOR(bytes.NewReader(val.Raw)); err != nil {
+		return err
+	}
+	idAddr, err := address.NewIDAddress(uint64(*id))
+	if err != nil {
+		return err
+	}
+	i.Added = append(i.Added, AddressPair{
+		ID: idAddr,
+		PK: pkAddr,
+	})
+	return nil
+}
+
+func (i *InitActorAddressChanges) Modify(key string, from, to *typegen.Deferred) error {
+	pkAddr, err := address.NewFromBytes([]byte(key))
+	if err != nil {
+		return err
+	}
+
+	fromID := new(typegen.CborInt)
+	if err := fromID.UnmarshalCBOR(bytes.NewReader(from.Raw)); err != nil {
+		return err
+	}
+	fromIDAddr, err := address.NewIDAddress(uint64(*fromID))
+	if err != nil {
+		return err
+	}
+
+	toID := new(typegen.CborInt)
+	if err := toID.UnmarshalCBOR(bytes.NewReader(to.Raw)); err != nil {
+		return err
+	}
+	toIDAddr, err := address.NewIDAddress(uint64(*toID))
+	if err != nil {
+		return err
+	}
+
+	i.Modified = append(i.Modified, AddressChange{
+		From: AddressPair{
+			ID: fromIDAddr,
+			PK: pkAddr,
+		},
+		To:   AddressPair{
+			ID: toIDAddr,
+			PK: pkAddr,
+		},
+	})
+	return nil
+}
+
+func (i *InitActorAddressChanges) Remove(key string, val *typegen.Deferred) error {
+	pkAddr, err := address.NewFromBytes([]byte(key))
+	if err != nil {
+		return err
+	}
+	id := new(typegen.CborInt)
+	if err := id.UnmarshalCBOR(bytes.NewReader(val.Raw)); err != nil {
+		return err
+	}
+	idAddr, err := address.NewIDAddress(uint64(*id))
+	if err != nil {
+		return err
+	}
+	i.Removed = append(i.Removed, AddressPair{
+		ID: idAddr,
+		PK: pkAddr,
+	})
+	return nil
+}
+
+func (sp *StatePredicates) OnAddressMapChange() DiffInitActorStateFunc {
+	return func(ctx context.Context, oldState, newState *init_.State) (changed bool, user UserData, err error) {
+		ctxStore := &contextStore{
+			ctx: ctx,
+			cst: sp.cst,
+		}
+
+		addressChanges := &InitActorAddressChanges{
+			Added:    []AddressPair{},
+			Modified: []AddressChange{},
+			Removed:  []AddressPair{},
+		}
+
+		if oldState.AddressMap.Equals(newState.AddressMap) {
+			return false, nil, nil
+		}
+
+		oldAddrs, err := adt.AsMap(ctxStore, oldState.AddressMap)
+		if err != nil {
+			return false, nil, err
+		}
+
+		newAddrs, err := adt.AsMap(ctxStore, newState.AddressMap)
+		if err != nil {
+			return false, nil, err
+		}
+
+		if err := DiffAdtMap(oldAddrs, newAddrs, addressChanges); err != nil {
+			return false, nil, err
+		}
+
+		if len(addressChanges.Added)+len(addressChanges.Removed)+len(addressChanges.Modified) == 0 {
+			return false, nil, nil
+		}
+
+		return true, addressChanges, nil
+	}
+}
+

cmd/lotus-chainwatch/processor/common_actors.go

@@ -1,23 +1,18 @@
 package processor
 
 import (
-	"bytes"
 	"context"
+	"fmt"
 	"time"
 
 	"golang.org/x/sync/errgroup"
 	"golang.org/x/xerrors"
 
 	"github.com/ipfs/go-cid"
-	typegen "github.com/whyrusleeping/cbor-gen"
 
 	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/lotus/chain/events/state"
 	"github.com/filecoin-project/specs-actors/actors/builtin"
-	_init "github.com/filecoin-project/specs-actors/actors/builtin/init"
-	"github.com/filecoin-project/specs-actors/actors/util/adt"
-
-	"github.com/filecoin-project/lotus/chain/types"
-	cw_util "github.com/filecoin-project/lotus/cmd/lotus-chainwatch/util"
 )
 
 func (p *Processor) setupCommonActors() error {
@@ -125,6 +120,11 @@ func (p *Processor) HandleCommonActorsChanges(ctx context.Context, actors map[ci
 	return grp.Wait()
 }
 
+type UpdateAddresses struct {
+	Old state.AddressPair
+	New state.AddressPair
+}
+
 func (p Processor) storeActorAddresses(ctx context.Context, actors map[cid.Cid]ActorTips) error {
 	start := time.Now()
 	defer func() {
@@ -141,41 +141,49 @@ func (p Processor) storeActorAddresses(ctx context.Context, actors map[cid.Cid]A
 	addressToID[builtin.StorageMarketActorAddr] = builtin.StorageMarketActorAddr
 	addressToID[builtin.VerifiedRegistryActorAddr] = builtin.VerifiedRegistryActorAddr
 	addressToID[builtin.BurntFundsActorAddr] = builtin.BurntFundsActorAddr
-	initActor, err := p.node.StateGetActor(ctx, builtin.InitActorAddr, types.EmptyTSK)
+
+	addressesToUpdate := []UpdateAddresses{}
+
+	pred := state.NewStatePredicates(p.node)
+	for _, act := range actors[builtin.InitActorCodeID] {
+		for _, info := range act {
+			changed, val, err := pred.OnInitActorChange(pred.OnAddressMapChange())(ctx, info.parentTsKey, info.tsKey)
+			if err != nil {
+				return err
+			}
+			if !changed {
+				continue
+			}
+			changes := val.(*state.InitActorAddressChanges)
+			for _, add := range changes.Added {
+				addressToID[add.PK] = add.ID
+			}
+			// we'll need to update any addresses that were modified, this indicates a reorg.
+			for _, mod := range changes.Modified {
+				addressesToUpdate = append(addressesToUpdate, UpdateAddresses{
+					Old: mod.From,
+					New: mod.To,
+				})
+			}
+		}
+	}
+
+	updateTx, err := p.db.Begin()
 	if err != nil {
 		return err
 	}
 
-	initActorRaw, err := p.node.ChainReadObj(ctx, initActor.Head)
+	for _, updates := range addressesToUpdate {
-	if err != nil {
+		if _, err := updateTx.Exec(
+			fmt.Sprintf("update id_address_map set id=%s, address=%s where id=%s and address=%s", updates.New.ID, updates.New.PK, updates.Old.ID, updates.Old.PK),
+			); err != nil {
+			return err
+		}
+	}
+	if err := updateTx.Commit(); err != nil {
 		return err
 	}
 
-	var initActorState _init.State
-	if err := initActorState.UnmarshalCBOR(bytes.NewReader(initActorRaw)); err != nil {
-		return err
-	}
-	ctxStore := cw_util.NewAPIIpldStore(ctx, p.node)
-	addrMap, err := adt.AsMap(ctxStore, initActorState.AddressMap)
-	if err != nil {
-		return err
-	}
-	// gross..
-	var actorID typegen.CborInt
-	if err := addrMap.ForEach(&actorID, func(key string) error {
-		longAddr, err := address.NewFromBytes([]byte(key))
-		if err != nil {
-			return err
-		}
-		shortAddr, err := address.NewIDAddress(uint64(actorID))
-		if err != nil {
-			return err
-		}
-		addressToID[longAddr] = shortAddr
-		return nil
-	}); err != nil {
-		return err
-	}
 	tx, err := p.db.Begin()
 	if err != nil {
 		return err
@@ -207,8 +215,7 @@ create temp table iam (like id_address_map excluding constraints) on commit drop
 		return err
 	}
 
-	// HACK until chain watch can handle reorgs we need to update this table when ID -> PubKey mappings change
+	if _, err := tx.Exec(`insert into id_address_map select * from iam on conflict (id) do nothing`); err != nil {
-	if _, err := tx.Exec(`insert into id_address_map select * from iam on conflict (id) do update set address = EXCLUDED.address`); err != nil {
 		return xerrors.Errorf("actor put: %w", err)
 	}