fix(rpc): fix processedEvents not handled if starting from pending (#2673)

Author: infrmtcs

blockchain/event_filter.go

@@ -124,9 +124,11 @@ func (e *EventFilter) Events(cToken *ContinuationToken, chunkSize uint64) ([]*Fi
 		}
 	}
 
+	var skippedEvents uint64
 	curBlock := e.fromBlock
 	// skip the blocks that we previously processed for this request
 	if cToken != nil {
+		skippedEvents = cToken.processedEvents
 		curBlock = cToken.fromBlock
 	}
 
@@ -161,14 +163,17 @@ func (e *EventFilter) Events(cToken *ContinuationToken, chunkSize uint64) ([]*Fi
 		}
 
 		var processedEvents uint64
-		matchedEvents, processedEvents, err = e.matcher.AppendBlockEvents(matchedEvents, header, receipts, cToken, chunkSize)
+		matchedEvents, processedEvents, err = e.matcher.AppendBlockEvents(matchedEvents, header, receipts, skippedEvents, chunkSize)
 		if err != nil {
 			if errors.Is(err, errChunkSizeReached) {
 				rToken = &ContinuationToken{fromBlock: curBlock, processedEvents: processedEvents}
 				break
 			}
 			return nil, nil, err
 		}
+
+		// Skipped events are processed, so we can reset the counter
+		skippedEvents = 0
 	}
 
 	if rToken == nil && remainingScannedBlocks == 0 && curBlock <= e.toBlock {

blockchain/event_matcher.go

@@ -93,7 +93,7 @@ func (e *EventMatcher) TestBloom(bloomFilter *bloom.BloomFilter) bool {
 }
 
 func (e *EventMatcher) AppendBlockEvents(matchedEventsSofar []*FilteredEvent, header *core.Header, receipts []*core.TransactionReceipt,
-	cToken *ContinuationToken, chunkSize uint64,
+	skippedEvents uint64, chunkSize uint64,
 ) ([]*FilteredEvent, uint64, error) {
 	processedEvents := uint64(0)
 	for _, receipt := range receipts {
@@ -106,7 +106,7 @@ func (e *EventMatcher) AppendBlockEvents(matchedEventsSofar []*FilteredEvent, he
 
 			// if last request was interrupted mid-block, and we are still processing that block, skip events
 			// that were already processed
-			if cToken != nil && header.Number == cToken.fromBlock && processedEvents < cToken.processedEvents {
+			if processedEvents < skippedEvents {
 				processedEvents++
 				continue
 			}

rpc/v6/events_test.go

@@ -205,4 +205,44 @@ func TestEvents(t *testing.T) {
 		assert.Nil(t, events.Events[0].BlockNumber)
 		assert.Equal(t, utils.HexToFelt(t, "0x785c2ada3f53fbc66078d47715c27718f92e6e48b96372b36e5197de69b82b5"), events.Events[0].TransactionHash)
 	})
+
+	t.Run("get pending events with pagination", func(t *testing.T) {
+		var err error
+		pendingB, err = gw.BlockByNumber(t.Context(), 5)
+		require.Nil(t, err)
+
+		args = rpc.EventsArg{
+			EventFilter: rpc.EventFilter{
+				FromBlock: &rpc.BlockID{Pending: true},
+				ToBlock:   &rpc.BlockID{Pending: true},
+			},
+			ResultPageRequest: rpc.ResultPageRequest{
+				ChunkSize: 1,
+			},
+		}
+
+		allEvents := []*core.Event{}
+
+		for _, receipt := range pendingB.Receipts {
+			allEvents = append(allEvents, receipt.Events...)
+		}
+
+		for i, expectedEvent := range allEvents {
+			events, err := handler.Events(args)
+			require.Nil(t, err)
+			require.Len(t, events.Events, 1)
+			actualEvent := events.Events[0]
+			if i == len(allEvents)-1 {
+				require.Empty(t, events.ContinuationToken)
+			} else {
+				require.NotEmpty(t, events.ContinuationToken)
+			}
+
+			assert.Equal(t, actualEvent.From, expectedEvent.From)
+			assert.Equal(t, actualEvent.Keys, expectedEvent.Keys)
+			assert.Equal(t, actualEvent.Data, expectedEvent.Data)
+
+			args.ContinuationToken = events.ContinuationToken
+		}
+	})
 }