Skip to content

Commit

Permalink
wip
Browse files Browse the repository at this point in the history
  • Loading branch information
@IronGauntlets
IronGauntlets committed Nov 22, 2024
1 parent d4111ba commit ae68ead
Show file tree
Hide file tree
Showing 2 changed files with 156 additions and 34 deletions.
17 changes: 14 additions & 3 deletions rpc/subscriptions.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package rpc
import (
"context"
"encoding/json"
"fmt"
"sync"

"github.com/NethermindEth/juno/blockchain"
Expand Down Expand Up @@ -81,11 +82,13 @@ func (h *Handler) SubscribeEvents(ctx context.Context, fromAddr *felt.Felt, keys
case <-subscriptionCtx.Done():
return
case header := <-headerSub.Recv():
fmt.Println("Calling process events from headerSub go routine")
h.processEvents(subscriptionCtx, w, id, header.Number, header.Number, fromAddr, keys)
}
}
}()

fmt.Println("Calling process events from from main sub go routine.")
h.processEvents(subscriptionCtx, w, id, requestedHeader.Number, headHeader.Number, fromAddr, keys)

wg.Wait()
Expand All @@ -95,24 +98,32 @@ func (h *Handler) SubscribeEvents(ctx context.Context, fromAddr *felt.Felt, keys
}

func (h *Handler) processEvents(ctx context.Context, w jsonrpc.Conn, id, from, to uint64, fromAddr *felt.Felt, keys [][]felt.Felt) {
fmt.Println("Inside process Events")
filter, err := h.bcReader.EventFilter(fromAddr, keys)
if err != nil {
h.log.Warnw("Error creating event filter", "err", err)
return
}
defer h.callAndLogErr(filter.Close, "Error closing event filter in events subscription")
fmt.Println("After h.bcReader.EventFilter")
defer func() {
fmt.Println("Returning from processEvents")
h.callAndLogErr(filter.Close, "Error closing event filter in events subscription")
}()

if err = setEventFilterRange(filter, &BlockID{Number: from}, &BlockID{Number: to}, to); err != nil {
h.log.Warnw("Error setting event filter range", "err", err)
return
}

var cToken *blockchain.ContinuationToken
filteredEvents, cToken, err := filter.Events(cToken, subscribeEventsChunkSize)
fmt.Println("Before calling Filter.Events")
filteredEvents, cToken, err := filter.Events(nil, subscribeEventsChunkSize)
fmt.Println("After Filter.Events1")
if err != nil {
fmt.Println("err is not nil--------")
h.log.Warnw("Error filtering events", "err", err)
return
}
fmt.Println("After Filter.Events2")

err = sendEvents(ctx, w, filteredEvents, id)
if err != nil {
Expand Down
173 changes: 142 additions & 31 deletions rpc/subscriptions_test.go
View file
Original file line number Diff line number Diff line change
@@ -1,24 +1,46 @@
package rpc_test
package rpc

import (
"context"
"fmt"
"encoding/json"
"io"
"net"
"testing"
"time"

"github.com/NethermindEth/juno/blockchain"
"github.com/NethermindEth/juno/clients/feeder"
"github.com/NethermindEth/juno/core"
"github.com/NethermindEth/juno/core/felt"
"github.com/NethermindEth/juno/feed"
"github.com/NethermindEth/juno/jsonrpc"
"github.com/NethermindEth/juno/mocks"
"github.com/NethermindEth/juno/rpc"
adaptfeeder "github.com/NethermindEth/juno/starknetdata/feeder"
"github.com/NethermindEth/juno/utils"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/mock/gomock"
)

// Due to the difference in how some test files in rpc use "package rpc" vs "package rpc_test" it was easiest to copy
// the fakeConn here.
// Todo: move all the subscription related test here
type fakeConn struct {
w io.Writer
}

func (fc *fakeConn) Write(p []byte) (int, error) {
return fc.w.Write(p)
}

func (fc *fakeConn) Equal(other jsonrpc.Conn) bool {
fc2, ok := other.(*fakeConn)
if !ok {
return false
}
return fc.w == fc2.w
}

func TestSubscribeEventsAndUnsubscribe(t *testing.T) {
log := utils.NewNopZapLogger()

Expand All @@ -28,7 +50,7 @@ func TestSubscribeEventsAndUnsubscribe(t *testing.T) {

mockChain := mocks.NewMockReader(mockCtrl)
mockSyncer := mocks.NewMockSyncReader(mockCtrl)
handler := rpc.New(mockChain, mockSyncer, nil, "", log)
handler := New(mockChain, mockSyncer, nil, "", log)

keys := make([][]felt.Felt, 1024+1)
fromAddr := new(felt.Felt).SetBytes([]byte("from_address"))
Expand All @@ -43,7 +65,7 @@ func TestSubscribeEventsAndUnsubscribe(t *testing.T) {

id, rpcErr := handler.SubscribeEvents(subCtx, fromAddr, keys, nil)
assert.Zero(t, id)
assert.Equal(t, rpc.ErrTooManyKeysInFilter, rpcErr)
assert.Equal(t, ErrTooManyKeysInFilter, rpcErr)
})

t.Run("Too many blocks back", func(t *testing.T) {
Expand All @@ -52,11 +74,11 @@ func TestSubscribeEventsAndUnsubscribe(t *testing.T) {

mockChain := mocks.NewMockReader(mockCtrl)
mockSyncer := mocks.NewMockSyncReader(mockCtrl)
handler := rpc.New(mockChain, mockSyncer, nil, "", log)
handler := New(mockChain, mockSyncer, nil, "", log)

keys := make([][]felt.Felt, 1)
fromAddr := new(felt.Felt).SetBytes([]byte("from_address"))
blockID := &rpc.BlockID{Number: 0}
blockID := &BlockID{Number: 0}

serverConn, clientConn := net.Pipe()
t.Cleanup(func() {
Expand All @@ -75,7 +97,7 @@ func TestSubscribeEventsAndUnsubscribe(t *testing.T) {

id, rpcErr := handler.SubscribeEvents(subCtx, fromAddr, keys, blockID)
assert.Zero(t, id)
assert.Equal(t, rpc.ErrTooManyBlocksBack, rpcErr)
assert.Equal(t, ErrTooManyBlocksBack, rpcErr)
})

t.Run("head is more than 1024", func(t *testing.T) {
Expand All @@ -84,25 +106,65 @@ func TestSubscribeEventsAndUnsubscribe(t *testing.T) {

id, rpcErr := handler.SubscribeEvents(subCtx, fromAddr, keys, blockID)
assert.Zero(t, id)
assert.Equal(t, rpc.ErrTooManyBlocksBack, rpcErr)
assert.Equal(t, ErrTooManyBlocksBack, rpcErr)
})
})

t.Run("Events from old blocks and new", func(t *testing.T) {
n := utils.Ptr(utils.Sepolia)
client := feeder.NewTestClient(t, n)
gw := adaptfeeder.New(client)
n := utils.Ptr(utils.Sepolia)
client := feeder.NewTestClient(t, n)
gw := adaptfeeder.New(client)

b1, err := gw.BlockByNumber(context.Background(), 56377)
require.NoError(t, err)

fromAddr := new(felt.Felt).SetBytes([]byte("some address"))
keys := [][]felt.Felt{{*new(felt.Felt).SetBytes([]byte("key1"))}}

filteredEvents := []*blockchain.FilteredEvent{
{
Event: b1.Receipts[0].Events[0],
BlockNumber: b1.Number,
BlockHash: new(felt.Felt).SetBytes([]byte("b1")),
TransactionHash: b1.Transactions[0].Hash(),
},
{
Event: b1.Receipts[1].Events[0],
BlockNumber: b1.Number + 1,
BlockHash: new(felt.Felt).SetBytes([]byte("b2")),
TransactionHash: b1.Transactions[1].Hash(),
},
}

var emittedEvents []*EmittedEvent
for _, e := range filteredEvents {
emittedEvents = append(emittedEvents, &EmittedEvent{
Event: &Event{
From: e.From,
Keys: e.Keys,
Data: e.Data,
},
BlockHash: e.BlockHash,
BlockNumber: &e.BlockNumber,
TransactionHash: e.TransactionHash,
})
}

b1, err := gw.BlockByNumber(context.Background(), 56377)
require.NoError(t, err)
t.Run("Events from old blocks", func(t *testing.T) {
mockCtrl := gomock.NewController(t)
t.Cleanup(mockCtrl.Finish)

// Make a shallow copy of b1 into b2 and b3. Then modify them accordingly.
b2, b3 := new(core.Block), new(core.Block)
b2.Header, b3.Header = new(core.Header), new(core.Header)
*b2.Header, *b3.Header = *b1.Header, *b1.Header
b2.Number = b1.Number + 1
b3.Number = b2.Number + 1
fmt.Println(b1.Number, b2.Number, b3.Number)
mockChain := mocks.NewMockReader(mockCtrl)
mockSyncer := mocks.NewMockSyncReader(mockCtrl)
mockEventFilterer := mocks.NewMockEventFilterer(mockCtrl)
handler := New(mockChain, mockSyncer, nil, "", log)

mockChain.EXPECT().HeadsHeader().Return(&core.Header{Number: b1.Number}, nil)
mockChain.EXPECT().BlockHeaderByNumber(b1.Number).Return(b1.Header, nil)
mockChain.EXPECT().EventFilter(fromAddr, keys).Return(mockEventFilterer, nil)

mockEventFilterer.EXPECT().SetRangeEndBlockByNumber(gomock.Any(), gomock.Any()).Return(nil).MaxTimes(2)
mockEventFilterer.EXPECT().Events(gomock.Any(), gomock.Any()).Return(filteredEvents, nil, nil)
mockEventFilterer.EXPECT().Close().AnyTimes()

serverConn, clientConn := net.Pipe()
t.Cleanup(func() {
Expand All @@ -111,25 +173,74 @@ func TestSubscribeEventsAndUnsubscribe(t *testing.T) {
})

subCtx := context.WithValue(context.Background(), jsonrpc.ConnKey{}, &fakeConn{w: serverConn})
fromAddr := b1.Receipts[0].Events[0].From
keys := make([][]felt.Felt, 1)
for _, k := range b1.Receipts[0].Events[0].Keys {
keys[0] = append(keys[0], *k)
id, rpcErr := handler.SubscribeEvents(subCtx, fromAddr, keys, &BlockID{Number: b1.Number})
require.Nil(t, rpcErr)

var marshalledResponses [][]byte
for _, e := range emittedEvents {
resp, err := marshalSubscriptionResponse(e, id.ID)
require.NoError(t, err)
marshalledResponses = append(marshalledResponses, resp)
}

for _, m := range marshalledResponses {
got := make([]byte, len(m))
_, err := clientConn.Read(got)
require.NoError(t, err)
assert.Equal(t, string(m), string(got))
}
})

t.Run("Events from new blocks", func(t *testing.T) {
mockCtrl := gomock.NewController(t)
t.Cleanup(mockCtrl.Finish)

mockChain := mocks.NewMockReader(mockCtrl)
mockSyncer := mocks.NewMockSyncReader(mockCtrl)
handler := rpc.New(mockChain, mockSyncer, nil, "", log)
mockEventFilterer := mocks.NewMockEventFilterer(mockCtrl)
handler := New(mockChain, mockSyncer, nil, "", log)

mockChain.EXPECT().HeadsHeader().Return(&core.Header{Number: b2.Number}, nil)
mockChain.EXPECT().BlockHeaderByNumber(b1.Number).Return(b1.Header, nil)
mockChain.EXPECT().HeadsHeader().Return(&core.Header{Number: b1.Number}, nil)
mockChain.EXPECT().EventFilter(fromAddr, keys).Return(mockEventFilterer, nil)

mockEventFilterer.EXPECT().SetRangeEndBlockByNumber(gomock.Any(), gomock.Any()).Return(nil).MaxTimes(2)
mockEventFilterer.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]*blockchain.FilteredEvent{filteredEvents[0]}, nil, nil)
mockEventFilterer.EXPECT().Close().AnyTimes()

serverConn, clientConn := net.Pipe()
t.Cleanup(func() {
require.NoError(t, serverConn.Close())
require.NoError(t, clientConn.Close())
})

_, rpcErr := handler.SubscribeEvents(subCtx, fromAddr, keys, &rpc.BlockID{Number: b1.Number})
subCtx := context.WithValue(context.Background(), jsonrpc.ConnKey{}, &fakeConn{w: serverConn})
id, rpcErr := handler.SubscribeEvents(subCtx, fromAddr, keys, nil)
require.Nil(t, rpcErr)

// Check from the conn that the correct id has been passed
headerFeed := feed.New[*core.Header]()
handler.newHeads = headerFeed

// headerFeed.Send()

resp, err := marshalSubscriptionResponse(emittedEvents[0], id.ID)
require.NoError(t, err)

got := make([]byte, len(resp))
_, err = clientConn.Read(got)
require.NoError(t, err)
assert.Equal(t, string(resp), string(got))

time.Sleep(100 * time.Millisecond)
})
}

func marshalSubscriptionResponse(e *EmittedEvent, id uint64) ([]byte, error) {
return json.Marshal(jsonrpc.Request{
Version: "2.0",
Method: "starknet_subscriptionEvents",
Params: map[string]any{
"subscription_id": id,
"result": e,
},
})
}

0 comments on commit ae68ead

Please sign in to comment.