cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge pull request #9808 from filecoin-project/issue/ref-fvm-1189

Browse Source 
Use EthHash compatible type for subscription and filter IDs
This commit is contained in:
Aayush Rajasekaran 2022-12-07 15:28:33 -05:00 committed by GitHub
commit cfceafb576
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 426 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
api/docgen/docgen.go
View File

@ -378,8 +378,11 @@ func init() {
ethFeeHistoryReward := [][]api.EthBigInt{} ethFeeHistoryReward := [][]api.EthBigInt{}
addExample(&ethFeeHistoryReward) addExample(&ethFeeHistoryReward)
addExample(api.EthFilterID("c5564560217c43e4bc0484df655e9019")) filterid, _ := api.EthHashFromHex("0x5CbEeC012345673f25E309Cc264f240bb0664031")
addExample(api.EthSubscriptionID("b62df77831484129adf6682332ad0725")) addExample(api.EthFilterID(filterid))
subid, _ := api.EthHashFromHex("0x5CbEeCF99d3fDB301234567c264f240bb0664031")
addExample(api.EthSubscriptionID(subid))
pstring := func(s string) *string { return &s } pstring := func(s string) *string { return &s }
addExample(&api.EthFilterSpec{ addExample(&api.EthFilterSpec{

5
api/eth_types.go
View File

@ -417,11 +417,10 @@ type EthFeeHistory struct {
Reward *[][]EthBigInt `json:"reward,omitempty"` Reward *[][]EthBigInt `json:"reward,omitempty"`
} }
// An opaque identifier generated by the Lotus node to refer to an installed filter. type EthFilterID EthHash
type EthFilterID string
// An opaque identifier generated by the Lotus node to refer to an active subscription. // An opaque identifier generated by the Lotus node to refer to an active subscription.
type EthSubscriptionID string type EthSubscriptionID EthHash
type EthFilterSpec struct { type EthFilterSpec struct {
// Interpreted as an epoch or one of "latest" for last mined block, "earliest" for first, // Interpreted as an epoch or one of "latest" for last mined block, "earliest" for first,

2
api/eth_types_test.go
View File

@ -176,7 +176,7 @@ func TestEthFilterResultMarshalJSON(t *testing.T) {
BlockHash: hash2, BlockHash: hash2,
BlockNumber: 53, BlockNumber: 53,
Topics: []EthBytes{hash1[:]}, Topics: []EthBytes{hash1[:]},
Data: hash1[:], Data: EthBytes(hash1[:]),
Address: addr, Address: addr,
} }
logjson, err := json.Marshal(log) logjson, err := json.Marshal(log)

BIN
build/openrpc/full.json.gz
View File

Binary file not shown.

19
chain/events/filter/event.go
View File

@ -7,7 +7,6 @@ import (
"sync" "sync"
"time" "time"
"github.com/google/uuid"
"github.com/ipfs/go-cid" "github.com/ipfs/go-cid"
cbg "github.com/whyrusleeping/cbor-gen" cbg "github.com/whyrusleeping/cbor-gen"
"golang.org/x/xerrors" "golang.org/x/xerrors"
@ -24,7 +23,7 @@ import (
const indexed uint8 = 0x01 const indexed uint8 = 0x01
type EventFilter struct { type EventFilter struct {
id string id types.FilterID
minHeight abi.ChainEpoch // minimum epoch to apply filter or -1 if no minimum minHeight abi.ChainEpoch // minimum epoch to apply filter or -1 if no minimum
maxHeight abi.ChainEpoch // maximum epoch to apply filter or -1 if no maximum maxHeight abi.ChainEpoch // maximum epoch to apply filter or -1 if no maximum
tipsetCid cid.Cid tipsetCid cid.Cid
@ -51,7 +50,7 @@ type CollectedEvent struct {
MsgCid cid.Cid // cid of message that produced event MsgCid cid.Cid // cid of message that produced event
} }
func (f *EventFilter) ID() string { func (f *EventFilter) ID() types.FilterID {
return f.id return f.id
} }
@ -290,7 +289,7 @@ type EventFilterManager struct {
EventIndex *EventIndex EventIndex *EventIndex
mu sync.Mutex // guards mutations to filters mu sync.Mutex // guards mutations to filters
filters map[string]*EventFilter filters map[types.FilterID]*EventFilter
currentHeight abi.ChainEpoch currentHeight abi.ChainEpoch
} }
@ -365,13 +364,13 @@ func (m *EventFilterManager) Install(ctx context.Context, minHeight, maxHeight a
return nil, xerrors.Errorf("historic event index disabled") return nil, xerrors.Errorf("historic event index disabled")
} }
id, err := uuid.NewRandom() id, err := newFilterID()
if err != nil { if err != nil {
return nil, xerrors.Errorf("new uuid: %w", err) return nil, xerrors.Errorf("new filter id: %w", err)
} }
f := &EventFilter{ f := &EventFilter{
id: id.String(), id: id,
minHeight: minHeight, minHeight: minHeight,
maxHeight: maxHeight, maxHeight: maxHeight,
tipsetCid: tipsetCid, tipsetCid: tipsetCid,
@ -389,15 +388,15 @@ func (m *EventFilterManager) Install(ctx context.Context, minHeight, maxHeight a
m.mu.Lock() m.mu.Lock()
if m.filters == nil { if m.filters == nil {
m.filters = make(map[string]*EventFilter) m.filters = make(map[types.FilterID]*EventFilter)
} }
m.filters[id.String()] = f m.filters[id] = f
m.mu.Unlock() m.mu.Unlock()
return f, nil return f, nil
} }
func (m *EventFilterManager) Remove(ctx context.Context, id string) error { func (m *EventFilterManager) Remove(ctx context.Context, id types.FilterID) error {
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
if _, found := m.filters[id]; !found { if _, found := m.filters[id]; !found {

19
chain/events/filter/mempool.go
View File

@ -5,7 +5,6 @@ import (
"sync" "sync"
"time" "time"
"github.com/google/uuid"
"github.com/ipfs/go-cid" "github.com/ipfs/go-cid"
"golang.org/x/xerrors" "golang.org/x/xerrors"
@ -14,7 +13,7 @@ import (
) )
type MemPoolFilter struct { type MemPoolFilter struct {
id string id types.FilterID
maxResults int // maximum number of results to collect, 0 is unlimited maxResults int // maximum number of results to collect, 0 is unlimited
ch chan<- interface{} ch chan<- interface{}
@ -25,7 +24,7 @@ type MemPoolFilter struct {
var _ Filter = (*MemPoolFilter)(nil) var _ Filter = (*MemPoolFilter)(nil)
func (f *MemPoolFilter) ID() string { func (f *MemPoolFilter) ID() types.FilterID {
return f.id return f.id
} }
@ -79,7 +78,7 @@ type MemPoolFilterManager struct {
MaxFilterResults int MaxFilterResults int
mu sync.Mutex // guards mutations to filters mu sync.Mutex // guards mutations to filters
filters map[string]*MemPoolFilter filters map[types.FilterID]*MemPoolFilter
} }
func (m *MemPoolFilterManager) WaitForMpoolUpdates(ctx context.Context, ch <-chan api.MpoolUpdate) { func (m *MemPoolFilterManager) WaitForMpoolUpdates(ctx context.Context, ch <-chan api.MpoolUpdate) {
@ -113,27 +112,27 @@ func (m *MemPoolFilterManager) processUpdate(ctx context.Context, u api.MpoolUpd
} }
func (m *MemPoolFilterManager) Install(ctx context.Context) (*MemPoolFilter, error) { func (m *MemPoolFilterManager) Install(ctx context.Context) (*MemPoolFilter, error) {
id, err := uuid.NewRandom() id, err := newFilterID()
if err != nil { if err != nil {
return nil, xerrors.Errorf("new uuid: %w", err) return nil, xerrors.Errorf("new filter id: %w", err)
} }
f := &MemPoolFilter{ f := &MemPoolFilter{
id: id.String(), id: id,
maxResults: m.MaxFilterResults, maxResults: m.MaxFilterResults,
} }
m.mu.Lock() m.mu.Lock()
if m.filters == nil { if m.filters == nil {
m.filters = make(map[string]*MemPoolFilter) m.filters = make(map[types.FilterID]*MemPoolFilter)
} }
m.filters[id.String()] = f m.filters[id] = f
m.mu.Unlock() m.mu.Unlock()
return f, nil return f, nil
} }
func (m *MemPoolFilterManager) Remove(ctx context.Context, id string) error { func (m *MemPoolFilterManager) Remove(ctx context.Context, id types.FilterID) error {
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
if _, found := m.filters[id]; !found { if _, found := m.filters[id]; !found {

29
chain/events/filter/store.go
View File

@ -5,10 +5,15 @@ import (
"errors" "errors"
"sync" "sync"
"time" "time"
"github.com/google/uuid"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/chain/types"
) )
type Filter interface { type Filter interface {
ID() string ID() types.FilterID
LastTaken() time.Time LastTaken() time.Time
SetSubChannel(chan<- interface{}) SetSubChannel(chan<- interface{})
ClearSubChannel() ClearSubChannel()
@ -16,8 +21,8 @@ type Filter interface {
type FilterStore interface { type FilterStore interface {
Add(context.Context, Filter) error Add(context.Context, Filter) error
Get(context.Context, string) (Filter, error) Get(context.Context, types.FilterID) (Filter, error)
Remove(context.Context, string) error Remove(context.Context, types.FilterID) error
NotTakenSince(when time.Time) []Filter // returns a list of filters that have not had their collected results taken NotTakenSince(when time.Time) []Filter // returns a list of filters that have not had their collected results taken
} }
@ -27,10 +32,20 @@ var (
ErrMaximumNumberOfFilters = errors.New("maximum number of filters registered") ErrMaximumNumberOfFilters = errors.New("maximum number of filters registered")
) )
func newFilterID() (types.FilterID, error) {
rawid, err := uuid.NewRandom()
if err != nil {
return types.FilterID{}, xerrors.Errorf("new uuid: %w", err)
}
id := types.FilterID{}
copy(id[:], rawid[:]) // uuid is 16 bytes, the last 16 bytes are zeroed
return id, nil
}
type memFilterStore struct { type memFilterStore struct {
max int max int
mu sync.Mutex mu sync.Mutex
filters map[string]Filter filters map[types.FilterID]Filter
} }
var _ FilterStore = (*memFilterStore)(nil) var _ FilterStore = (*memFilterStore)(nil)
@ -38,7 +53,7 @@ var _ FilterStore = (*memFilterStore)(nil)
func NewMemFilterStore(maxFilters int) FilterStore { func NewMemFilterStore(maxFilters int) FilterStore {
return &memFilterStore{ return &memFilterStore{
max: maxFilters, max: maxFilters,
filters: make(map[string]Filter), filters: make(map[types.FilterID]Filter),
} }
} }
@ -57,7 +72,7 @@ func (m *memFilterStore) Add(_ context.Context, f Filter) error {
return nil return nil
} }
func (m *memFilterStore) Get(_ context.Context, id string) (Filter, error) { func (m *memFilterStore) Get(_ context.Context, id types.FilterID) (Filter, error) {
m.mu.Lock() m.mu.Lock()
f, found := m.filters[id] f, found := m.filters[id]
m.mu.Unlock() m.mu.Unlock()
@ -67,7 +82,7 @@ func (m *memFilterStore) Get(_ context.Context, id string) (Filter, error) {
return f, nil return f, nil
} }
func (m *memFilterStore) Remove(_ context.Context, id string) error { func (m *memFilterStore) Remove(_ context.Context, id types.FilterID) error {
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()

19
chain/events/filter/tipset.go
View File

@ -5,14 +5,13 @@ import (
"sync" "sync"
"time" "time"
"github.com/google/uuid"
"golang.org/x/xerrors" "golang.org/x/xerrors"
"github.com/filecoin-project/lotus/chain/types" "github.com/filecoin-project/lotus/chain/types"
) )
type TipSetFilter struct { type TipSetFilter struct {
id string id types.FilterID
maxResults int // maximum number of results to collect, 0 is unlimited maxResults int // maximum number of results to collect, 0 is unlimited
ch chan<- interface{} ch chan<- interface{}
@ -23,7 +22,7 @@ type TipSetFilter struct {
var _ Filter = (*TipSetFilter)(nil) var _ Filter = (*TipSetFilter)(nil)
func (f *TipSetFilter) ID() string { func (f *TipSetFilter) ID() types.FilterID {
return f.id return f.id
} }
@ -77,7 +76,7 @@ type TipSetFilterManager struct {
MaxFilterResults int MaxFilterResults int
mu sync.Mutex // guards mutations to filters mu sync.Mutex // guards mutations to filters
filters map[string]*TipSetFilter filters map[types.FilterID]*TipSetFilter
} }
func (m *TipSetFilterManager) Apply(ctx context.Context, from, to *types.TipSet) error { func (m *TipSetFilterManager) Apply(ctx context.Context, from, to *types.TipSet) error {
@ -100,27 +99,27 @@ func (m *TipSetFilterManager) Revert(ctx context.Context, from, to *types.TipSet
} }
func (m *TipSetFilterManager) Install(ctx context.Context) (*TipSetFilter, error) { func (m *TipSetFilterManager) Install(ctx context.Context) (*TipSetFilter, error) {
id, err := uuid.NewRandom() id, err := newFilterID()
if err != nil { if err != nil {
return nil, xerrors.Errorf("new uuid: %w", err) return nil, xerrors.Errorf("new filter id: %w", err)
} }
f := &TipSetFilter{ f := &TipSetFilter{
id: id.String(), id: id,
maxResults: m.MaxFilterResults, maxResults: m.MaxFilterResults,
} }
m.mu.Lock() m.mu.Lock()
if m.filters == nil { if m.filters == nil {
m.filters = make(map[string]*TipSetFilter) m.filters = make(map[types.FilterID]*TipSetFilter)
} }
m.filters[id.String()] = f m.filters[id] = f
m.mu.Unlock() m.mu.Unlock()
return f, nil return f, nil
} }
func (m *TipSetFilterManager) Remove(ctx context.Context, id string) error { func (m *TipSetFilterManager) Remove(ctx context.Context, id types.FilterID) error {
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
if _, found := m.filters[id]; !found { if _, found := m.filters[id]; !found {

2
chain/types/event.go
View File

@ -22,3 +22,5 @@ type EventEntry struct {
// Any DAG-CBOR encodeable type. // Any DAG-CBOR encodeable type.
Value []byte Value []byte
} }
type FilterID [32]byte // compatible with EthHash

289
documentation/en/api-v1-unstable-methods.md
View File

@ -2510,7 +2510,40 @@ Perms: write
Inputs: Inputs:
```json ```json
[ [
"c5564560217c43e4bc0484df655e9019" [
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
92,
190,
236,
1,
35,
69,
103,
63,
37,
227,
9,
204,
38,
79,
36,
11,
176,
102,
64,
49
]
] ]
``` ```
@ -2531,7 +2564,40 @@ Perms: write
Inputs: Inputs:
```json ```json
[ [
"c5564560217c43e4bc0484df655e9019" [
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
92,
190,
236,
1,
35,
69,
103,
63,
37,
227,
9,
204,
38,
79,
36,
11,
176,
102,
64,
49
]
] ]
``` ```
@ -2769,7 +2835,43 @@ Perms: write
Inputs: `null` Inputs: `null`
Response: `"c5564560217c43e4bc0484df655e9019"` Response:
```json
[
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
92,
190,
236,
1,
35,
69,
103,
63,
37,
227,
9,
204,
38,
79,
36,
11,
176,
102,
64,
49
]
```
### EthNewFilter ### EthNewFilter
Installs a persistent filter based on given filter spec. Installs a persistent filter based on given filter spec.
@ -2790,7 +2892,43 @@ Inputs:
] ]
``` ```
Response: `"c5564560217c43e4bc0484df655e9019"` Response:
```json
[
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
92,
190,
236,
1,
35,
69,
103,
63,
37,
227,
9,
204,
38,
79,
36,
11,
176,
102,
64,
49
]
```
### EthNewPendingTransactionFilter ### EthNewPendingTransactionFilter
Installs a persistent filter to notify when new messages arrive in the message pool. Installs a persistent filter to notify when new messages arrive in the message pool.
@ -2800,7 +2938,43 @@ Perms: write
Inputs: `null` Inputs: `null`
Response: `"c5564560217c43e4bc0484df655e9019"` Response:
```json
[
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
92,
190,
236,
1,
35,
69,
103,
63,
37,
227,
9,
204,
38,
79,
36,
11,
176,
102,
64,
49
]
```
### EthProtocolVersion ### EthProtocolVersion
@ -2854,7 +3028,40 @@ Inputs:
Response: Response:
```json ```json
{ {
"subscription": "b62df77831484129adf6682332ad0725", "subscription": [
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
92,
190,
236,
249,
157,
63,
219,
48,
18,
52,
86,
124,
38,
79,
36,
11,
176,
102,
64,
49
],
"result": {} "result": {}
} }
``` ```
@ -2868,7 +3075,40 @@ Perms: write
Inputs: Inputs:
```json ```json
[ [
"c5564560217c43e4bc0484df655e9019" [
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
92,
190,
236,
1,
35,
69,
103,
63,
37,
227,
9,
204,
38,
79,
36,
11,
176,
102,
64,
49
]
] ]
``` ```
@ -2883,7 +3123,40 @@ Perms: write
Inputs: Inputs:
```json ```json
[ [
"b62df77831484129adf6682332ad0725" [
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
92,
190,
236,
249,
157,
63,
219,
48,
18,
52,
86,
124,
38,
79,
36,
11,
176,
102,
64,
49
]
] ]
``` ```

101
itests/eth_filter_test.go
View File

@ -294,11 +294,11 @@ func TestEthNewFilterCatchAll(t *testing.T) {
// expect to have seen iteration number of events // expect to have seen iteration number of events
require.Equal(iterations, len(res.Results)) require.Equal(iterations, len(res.Results))
topic1Hash := api.EthHashData([]byte{0x42, 0x11, 0x11}) topic1 := api.EthBytes(leftpad32([]byte{0x11, 0x11}))
topic2Hash := api.EthHashData([]byte{0x42, 0x22, 0x22}) topic2 := api.EthBytes(leftpad32([]byte{0x22, 0x22}))
topic3Hash := api.EthHashData([]byte{0x42, 0x33, 0x33}) topic3 := api.EthBytes(leftpad32([]byte{0x33, 0x33}))
topic4Hash := api.EthHashData([]byte{0x42, 0x44, 0x44}) topic4 := api.EthBytes(leftpad32([]byte{0x44, 0x44}))
data1Hash := api.EthHashData([]byte{0x48, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88}) data1 := api.EthBytes(leftpad32([]byte{0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88}))
for _, r := range res.Results { for _, r := range res.Results {
// since response is a union and Go doesn't support them well, go-jsonrpc won't give us typed results // since response is a union and Go doesn't support them well, go-jsonrpc won't give us typed results
@ -323,13 +323,12 @@ func TestEthNewFilterCatchAll(t *testing.T) {
require.Equal(tsCidHash, elog.BlockHash, "block hash") require.Equal(tsCidHash, elog.BlockHash, "block hash")
require.Equal(4, len(elog.Topics), "number of topics") require.Equal(4, len(elog.Topics), "number of topics")
require.Equal(topic1Hash, elog.Topics[0], "topic1") require.Equal(topic1, elog.Topics[0], "topic1")
require.Equal(topic2Hash, elog.Topics[1], "topic2") require.Equal(topic2, elog.Topics[1], "topic2")
require.Equal(topic3Hash, elog.Topics[2], "topic3") require.Equal(topic3, elog.Topics[2], "topic3")
require.Equal(topic4Hash, elog.Topics[3], "topic4") require.Equal(topic4, elog.Topics[3], "topic4")
require.Equal(1, len(elog.Data), "number of data") require.Equal(data1, elog.Data, "data1")
require.Equal(data1Hash, elog.Data[0], "data1")
} }
} }
@ -368,67 +367,76 @@ func ParseEthLog(in map[string]interface{}) (*api.EthLog, error) {
} }
s, ok := v.(string) s, ok := v.(string)
if !ok { if !ok {
return nil, xerrors.Errorf(k + " not a string") return nil, xerrors.Errorf(k + ": not a string")
} }
el.Removed, err = strconv.ParseBool(s) el.Removed, err = strconv.ParseBool(s)
if err != nil { if err != nil {
return nil, err return nil, xerrors.Errorf("%s: %w", k, err)
} }
case "address": case "address":
s, ok := v.(string) s, ok := v.(string)
if !ok { if !ok {
return nil, xerrors.Errorf(k + " not a string") return nil, xerrors.Errorf(k + ": not a string")
} }
el.Address, err = api.EthAddressFromHex(s) el.Address, err = api.EthAddressFromHex(s)
if err != nil { if err != nil {
return nil, err return nil, xerrors.Errorf("%s: %w", k, err)
} }
case "logIndex": case "logIndex":
el.LogIndex, err = ethUint64(k, v) el.LogIndex, err = ethUint64(k, v)
if err != nil { if err != nil {
return nil, err return nil, xerrors.Errorf("%s: %w", k, err)
} }
case "transactionIndex": case "transactionIndex":
el.TransactionIndex, err = ethUint64(k, v) el.TransactionIndex, err = ethUint64(k, v)
if err != nil { if err != nil {
return nil, err return nil, xerrors.Errorf("%s: %w", k, err)
} }
case "blockNumber": case "blockNumber":
el.BlockNumber, err = ethUint64(k, v) el.BlockNumber, err = ethUint64(k, v)
if err != nil { if err != nil {
return nil, err return nil, xerrors.Errorf("%s: %w", k, err)
} }
case "transactionHash": case "transactionHash":
el.TransactionHash, err = ethHash(k, v) el.TransactionHash, err = ethHash(k, v)
if err != nil { if err != nil {
return nil, err return nil, xerrors.Errorf("%s: %w", k, err)
} }
case "blockHash": case "blockHash":
el.BlockHash, err = ethHash(k, v) el.BlockHash, err = ethHash(k, v)
if err != nil { if err != nil {
return nil, err return nil, xerrors.Errorf("%s: %w", k, err)
} }
case "data": case "data":
sl, ok := v.([]interface{}) s, ok := v.(string)
if !ok { if !ok {
return nil, xerrors.Errorf(k + " not a slice") return nil, xerrors.Errorf(k + ": not a string")
} }
for _, s := range sl { data, err := hex.DecodeString(s[2:])
data, err := hex.DecodeString(s.(string)) if err != nil {
if err != nil { return nil, xerrors.Errorf("%s: %w", k, err)
return nil, err
}
el.Data = data
} }
el.Data = data
case "topics": case "topics":
s, ok := v.(string)
if ok {
topic, err := hex.DecodeString(s[2:])
if err != nil {
return nil, xerrors.Errorf("%s: %w", k, err)
}
el.Topics = append(el.Topics, topic)
continue
}
sl, ok := v.([]interface{}) sl, ok := v.([]interface{})
if !ok { if !ok {
return nil, xerrors.Errorf(k + " not a slice") return nil, xerrors.Errorf(k + ": not a slice")
} }
for _, s := range sl { for _, s := range sl {
topic, err := hex.DecodeString(s.(string)) topic, err := hex.DecodeString(s.(string)[2:])
if err != nil { if err != nil {
return nil, err return nil, xerrors.Errorf("%s: %w", k, err)
} }
el.Topics = append(el.Topics, topic) el.Topics = append(el.Topics, topic)
} }
@ -542,11 +550,11 @@ func TestEthGetLogsAll(t *testing.T) {
ethContractAddr, err := api.EthAddressFromFilecoinAddress(*actor.Address) ethContractAddr, err := api.EthAddressFromFilecoinAddress(*actor.Address)
require.NoError(err) require.NoError(err)
topic1Hash := api.EthHashData([]byte{0x42, 0x11, 0x11}) topic1 := api.EthBytes(leftpad32([]byte{0x11, 0x11}))
topic2Hash := api.EthHashData([]byte{0x42, 0x22, 0x22}) topic2 := api.EthBytes(leftpad32([]byte{0x22, 0x22}))
topic3Hash := api.EthHashData([]byte{0x42, 0x33, 0x33}) topic3 := api.EthBytes(leftpad32([]byte{0x33, 0x33}))
topic4Hash := api.EthHashData([]byte{0x42, 0x44, 0x44}) topic4 := api.EthBytes(leftpad32([]byte{0x44, 0x44}))
data1Hash := api.EthHashData([]byte{0x48, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88}) data1 := api.EthBytes(leftpad32([]byte{0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88}))
pstring := func(s string) *string { return &s } pstring := func(s string) *string { return &s }
@ -582,13 +590,12 @@ func TestEthGetLogsAll(t *testing.T) {
require.Equal(tsCidHash, elog.BlockHash, "block hash") require.Equal(tsCidHash, elog.BlockHash, "block hash")
require.Equal(4, len(elog.Topics), "number of topics") require.Equal(4, len(elog.Topics), "number of topics")
require.Equal(topic1Hash, elog.Topics[0], "topic1") require.Equal(topic1, elog.Topics[0], "topic1")
require.Equal(topic2Hash, elog.Topics[1], "topic2") require.Equal(topic2, elog.Topics[1], "topic2")
require.Equal(topic3Hash, elog.Topics[2], "topic3") require.Equal(topic3, elog.Topics[2], "topic3")
require.Equal(topic4Hash, elog.Topics[3], "topic4") require.Equal(topic4, elog.Topics[3], "topic4")
require.Equal(1, len(elog.Data), "number of data") require.Equal(data1, elog.Data, "data1")
require.Equal(data1Hash, elog.Data[0], "data1")
} }
} }
@ -706,3 +713,13 @@ func TestEthSubscribeLogs(t *testing.T) {
// expect to have seen all logs // expect to have seen all logs
require.Equal(len(received), len(subResponses)) require.Equal(len(received), len(subResponses))
} }
func leftpad32(orig []byte) []byte {
needed := 32 - len(orig)
if needed <= 0 {
return orig
}
ret := make([]byte, 32)
copy(ret[needed:], orig)
return ret
}

52
node/impl/full/eth.go
View File

@ -696,7 +696,7 @@ func (e *EthEvent) EthGetFilterChanges(ctx context.Context, id api.EthFilterID)
return nil, api.ErrNotSupported return nil, api.ErrNotSupported
} }
f, err := e.FilterStore.Get(ctx, string(id)) f, err := e.FilterStore.Get(ctx, types.FilterID(id))
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -718,7 +718,7 @@ func (e *EthEvent) EthGetFilterLogs(ctx context.Context, id api.EthFilterID) (*a
return nil, api.ErrNotSupported return nil, api.ErrNotSupported
} }
f, err := e.FilterStore.Get(ctx, string(id)) f, err := e.FilterStore.Get(ctx, types.FilterID(id))
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -823,22 +823,22 @@ func (e *EthEvent) installEthFilterSpec(ctx context.Context, filterSpec *api.Eth
func (e *EthEvent) EthNewFilter(ctx context.Context, filterSpec *api.EthFilterSpec) (api.EthFilterID, error) { func (e *EthEvent) EthNewFilter(ctx context.Context, filterSpec *api.EthFilterSpec) (api.EthFilterID, error) {
if e.FilterStore == nil || e.EventFilterManager == nil { if e.FilterStore == nil || e.EventFilterManager == nil {
return "", api.ErrNotSupported return api.EthFilterID{}, api.ErrNotSupported
} }
f, err := e.installEthFilterSpec(ctx, filterSpec) f, err := e.installEthFilterSpec(ctx, filterSpec)
if err != nil { if err != nil {
return "", err return api.EthFilterID{}, err
} }
if err := e.FilterStore.Add(ctx, f); err != nil { if err := e.FilterStore.Add(ctx, f); err != nil {
// Could not record in store, attempt to delete filter to clean up // Could not record in store, attempt to delete filter to clean up
err2 := e.TipSetFilterManager.Remove(ctx, f.ID()) err2 := e.TipSetFilterManager.Remove(ctx, f.ID())
if err2 != nil { if err2 != nil {
return "", xerrors.Errorf("encountered error %v while removing new filter due to %v", err2, err) return api.EthFilterID{}, xerrors.Errorf("encountered error %v while removing new filter due to %v", err2, err)
} }
return "", err return api.EthFilterID{}, err
} }
return api.EthFilterID(f.ID()), nil return api.EthFilterID(f.ID()), nil
@ -846,22 +846,22 @@ func (e *EthEvent) EthNewFilter(ctx context.Context, filterSpec *api.EthFilterSp
func (e *EthEvent) EthNewBlockFilter(ctx context.Context) (api.EthFilterID, error) { func (e *EthEvent) EthNewBlockFilter(ctx context.Context) (api.EthFilterID, error) {
if e.FilterStore == nil || e.TipSetFilterManager == nil { if e.FilterStore == nil || e.TipSetFilterManager == nil {
return "", api.ErrNotSupported return api.EthFilterID{}, api.ErrNotSupported
} }
f, err := e.TipSetFilterManager.Install(ctx) f, err := e.TipSetFilterManager.Install(ctx)
if err != nil { if err != nil {
return "", err return api.EthFilterID{}, err
} }
if err := e.FilterStore.Add(ctx, f); err != nil { if err := e.FilterStore.Add(ctx, f); err != nil {
// Could not record in store, attempt to delete filter to clean up // Could not record in store, attempt to delete filter to clean up
err2 := e.TipSetFilterManager.Remove(ctx, f.ID()) err2 := e.TipSetFilterManager.Remove(ctx, f.ID())
if err2 != nil { if err2 != nil {
return "", xerrors.Errorf("encountered error %v while removing new filter due to %v", err2, err) return api.EthFilterID{}, xerrors.Errorf("encountered error %v while removing new filter due to %v", err2, err)
} }
return "", err return api.EthFilterID{}, err
} }
return api.EthFilterID(f.ID()), nil return api.EthFilterID(f.ID()), nil
@ -869,22 +869,22 @@ func (e *EthEvent) EthNewBlockFilter(ctx context.Context) (api.EthFilterID, erro
func (e *EthEvent) EthNewPendingTransactionFilter(ctx context.Context) (api.EthFilterID, error) { func (e *EthEvent) EthNewPendingTransactionFilter(ctx context.Context) (api.EthFilterID, error) {
if e.FilterStore == nil || e.MemPoolFilterManager == nil { if e.FilterStore == nil || e.MemPoolFilterManager == nil {
return "", api.ErrNotSupported return api.EthFilterID{}, api.ErrNotSupported
} }
f, err := e.MemPoolFilterManager.Install(ctx) f, err := e.MemPoolFilterManager.Install(ctx)
if err != nil { if err != nil {
return "", err return api.EthFilterID{}, err
} }
if err := e.FilterStore.Add(ctx, f); err != nil { if err := e.FilterStore.Add(ctx, f); err != nil {
// Could not record in store, attempt to delete filter to clean up // Could not record in store, attempt to delete filter to clean up
err2 := e.MemPoolFilterManager.Remove(ctx, f.ID()) err2 := e.MemPoolFilterManager.Remove(ctx, f.ID())
if err2 != nil { if err2 != nil {
return "", xerrors.Errorf("encountered error %v while removing new filter due to %v", err2, err) return api.EthFilterID{}, xerrors.Errorf("encountered error %v while removing new filter due to %v", err2, err)
} }
return "", err return api.EthFilterID{}, err
} }
return api.EthFilterID(f.ID()), nil return api.EthFilterID(f.ID()), nil
@ -895,7 +895,7 @@ func (e *EthEvent) EthUninstallFilter(ctx context.Context, id api.EthFilterID) (
return false, api.ErrNotSupported return false, api.ErrNotSupported
} }
f, err := e.FilterStore.Get(ctx, string(id)) f, err := e.FilterStore.Get(ctx, types.FilterID(id))
if err != nil { if err != nil {
if errors.Is(err, filter.ErrFilterNotFound) { if errors.Is(err, filter.ErrFilterNotFound) {
return false, nil return false, nil
@ -956,7 +956,7 @@ func (e *EthEvent) EthSubscribe(ctx context.Context, eventType string, params *a
f, err := e.TipSetFilterManager.Install(ctx) f, err := e.TipSetFilterManager.Install(ctx)
if err != nil { if err != nil {
// clean up any previous filters added and stop the sub // clean up any previous filters added and stop the sub
_, _ = e.EthUnsubscribe(ctx, api.EthSubscriptionID(sub.id)) _, _ = e.EthUnsubscribe(ctx, sub.id)
return nil, err return nil, err
} }
sub.addFilter(ctx, f) sub.addFilter(ctx, f)
@ -978,7 +978,7 @@ func (e *EthEvent) EthSubscribe(ctx context.Context, eventType string, params *a
f, err := e.EventFilterManager.Install(ctx, -1, -1, cid.Undef, []address.Address{}, keys) f, err := e.EventFilterManager.Install(ctx, -1, -1, cid.Undef, []address.Address{}, keys)
if err != nil { if err != nil {
// clean up any previous filters added and stop the sub // clean up any previous filters added and stop the sub
_, _ = e.EthUnsubscribe(ctx, api.EthSubscriptionID(sub.id)) _, _ = e.EthUnsubscribe(ctx, sub.id)
return nil, err return nil, err
} }
sub.addFilter(ctx, f) sub.addFilter(ctx, f)
@ -994,7 +994,7 @@ func (e *EthEvent) EthUnsubscribe(ctx context.Context, id api.EthSubscriptionID)
return false, api.ErrNotSupported return false, api.ErrNotSupported
} }
filters, err := e.SubManager.StopSubscription(ctx, string(id)) filters, err := e.SubManager.StopSubscription(ctx, id)
if err != nil { if err != nil {
return false, nil return false, nil
} }
@ -1130,14 +1130,16 @@ type EthSubscriptionManager struct {
StateAPI StateAPI StateAPI StateAPI
ChainAPI ChainAPI ChainAPI ChainAPI
mu sync.Mutex mu sync.Mutex
subs map[string]*ethSubscription subs map[api.EthSubscriptionID]*ethSubscription
} }
func (e *EthSubscriptionManager) StartSubscription(ctx context.Context) (*ethSubscription, error) { // nolint func (e *EthSubscriptionManager) StartSubscription(ctx context.Context) (*ethSubscription, error) { // nolint
id, err := uuid.NewRandom() rawid, err := uuid.NewRandom()
if err != nil { if err != nil {
return nil, xerrors.Errorf("new uuid: %w", err) return nil, xerrors.Errorf("new uuid: %w", err)
} }
id := api.EthSubscriptionID{}
copy(id[:], rawid[:]) // uuid is 16 bytes
ctx, quit := context.WithCancel(ctx) ctx, quit := context.WithCancel(ctx)
@ -1145,7 +1147,7 @@ func (e *EthSubscriptionManager) StartSubscription(ctx context.Context) (*ethSub
Chain: e.Chain, Chain: e.Chain,
StateAPI: e.StateAPI, StateAPI: e.StateAPI,
ChainAPI: e.ChainAPI, ChainAPI: e.ChainAPI,
id: id.String(), id: id,
in: make(chan interface{}, 200), in: make(chan interface{}, 200),
out: make(chan api.EthSubscriptionResponse, 20), out: make(chan api.EthSubscriptionResponse, 20),
quit: quit, quit: quit,
@ -1153,7 +1155,7 @@ func (e *EthSubscriptionManager) StartSubscription(ctx context.Context) (*ethSub
e.mu.Lock() e.mu.Lock()
if e.subs == nil { if e.subs == nil {
e.subs = make(map[string]*ethSubscription) e.subs = make(map[api.EthSubscriptionID]*ethSubscription)
} }
e.subs[sub.id] = sub e.subs[sub.id] = sub
e.mu.Unlock() e.mu.Unlock()
@ -1163,7 +1165,7 @@ func (e *EthSubscriptionManager) StartSubscription(ctx context.Context) (*ethSub
return sub, nil return sub, nil
} }
func (e *EthSubscriptionManager) StopSubscription(ctx context.Context, id string) ([]filter.Filter, error) { func (e *EthSubscriptionManager) StopSubscription(ctx context.Context, id api.EthSubscriptionID) ([]filter.Filter, error) {
e.mu.Lock() e.mu.Lock()
defer e.mu.Unlock() defer e.mu.Unlock()
@ -1181,7 +1183,7 @@ type ethSubscription struct {
Chain *store.ChainStore Chain *store.ChainStore
StateAPI StateAPI StateAPI StateAPI
ChainAPI ChainAPI ChainAPI ChainAPI
id string id api.EthSubscriptionID
in chan interface{} in chan interface{}
out chan api.EthSubscriptionResponse out chan api.EthSubscriptionResponse
@ -1205,7 +1207,7 @@ func (e *ethSubscription) start(ctx context.Context) {
return return
case v := <-e.in: case v := <-e.in:
resp := api.EthSubscriptionResponse{ resp := api.EthSubscriptionResponse{
SubscriptionID: api.EthSubscriptionID(e.id), SubscriptionID: e.id,
} }
var err error var err error