ethmonitor: LogsStream

Author: pkieltyka

TODO.md

@@ -0,0 +1,48 @@
+TODO
+====
+
+ethtxn and ethwallet improvements .. naming etc.
+
+------
+
+1. Look at some of the viem and ox terminology.. "provider" ..?
+2. rename..? ethrpc.Client and ethrpc.Provider .. and .NewClient .. its a bit confusing
+    .... lets ask the AI its opinion..
+
+
+------
+
+
+
+1. ethreceipts using AsMessage .. dont need to, see note in receipt.go
+2. search "TODO" in ethkit..
+
+---
+
+
+4. ethrpc, pass in a json marshaller, would be better.. 
+5. review ethmonitor for the logs and streaming stuff, to build blocks from
+subscriptions, etc.
+
+6. any other cleanup in ethkit I should do...? lets make a list anyways..
+
+
+
+
+
+--------
+
+IMPROVEMENTS
+============
+
+1. ethmonitor: newHeads and newLogs .. can we use these..?
+2. ethreceipts: review TODO's to make it more efficient -- not worth it right now..
+3. ethrpc: remove pkg-level methods and pass in optional jsonmarshaller/unmarshaller -- have an idea how to do it..
+4. ethrpc: rename interface.go -- yes, at the right time..
+
+
+---------
+
+ethmonitor..
+
+1. lets adapt chain-newheads to see how "newHeads" looks..?

cmd/chain-watch/main.go

@@ -74,6 +74,11 @@ func main() {
 	chainID, _ := provider.ChainID(context.Background())
 	fmt.Println("=> chain id:", chainID.String())
 
+	ls := ethmonitor.NewLogsStream(provider)
+	ls.Do(context.Background())
+
+	return
+
 	// Monitor options
 	cachestore.MaxKeyLength = 180
 	monitorOptions := ethmonitor.DefaultOptions

ethmonitor/ethmonitor.go

@@ -524,7 +524,7 @@ func (m *Monitor) monitor() error {
 			}
 
 			// fetch the next block, either via the stream or via a poll
-			nextBlock, nextBlockPayload, miss, err := m.fetchNextBlock(ctx)
+			nextBlock, _, miss, err := m.fetchNextBlock(ctx)
 			if err != nil {
 				if errors.Is(err, context.DeadlineExceeded) {
 					m.log.Info(fmt.Sprintf("ethmonitor: fetchNextBlock timed out: '%v', for blockNum:%v, retrying..", err, m.nextBlockNumber))
@@ -546,7 +546,7 @@ func (m *Monitor) monitor() error {
 			}
 
 			// build deterministic set of add/remove events which construct the canonical chain
-			events, err = m.buildCanonicalChain(ctx, nextBlock, nextBlockPayload, events)
+			events, err = m.buildCanonicalChain(ctx, nextBlock, events)
 			if err != nil {
 				m.log.Warn(fmt.Sprintf("ethmonitor: error reported '%v', failed to build chain for next blockNum:%d blockHash:%s, retrying..",
 					err, nextBlock.NumberU64(), nextBlock.Hash().Hex()))
@@ -582,7 +582,7 @@ func (m *Monitor) monitor() error {
 	}
 }
 
-func (m *Monitor) buildCanonicalChain(ctx context.Context, nextBlock *types.Block, nextBlockPayload []byte, events Blocks) (Blocks, error) {
+func (m *Monitor) buildCanonicalChain(ctx context.Context, nextBlock *types.Block, events Blocks) (Blocks, error) {
 	select {
 	case <-ctx.Done():
 		return nil, ctx.Err()
@@ -629,13 +629,13 @@ func (m *Monitor) buildCanonicalChain(ctx context.Context, nextBlock *types.Bloc
 	time.Sleep(pause)
 
 	// Fetch/connect the broken chain backwards by traversing recursively via parent hashes
-	nextParentBlock, nextParentBlockPayload, err := m.fetchBlockByHash(ctx, nextBlock.ParentHash())
+	nextParentBlock, _, err := m.fetchBlockByHash(ctx, nextBlock.ParentHash())
 	if err != nil {
 		// NOTE: this is okay, it will auto-retry
 		return events, err
 	}
 
-	events, err = m.buildCanonicalChain(ctx, nextParentBlock, nextParentBlockPayload, events)
+	events, err = m.buildCanonicalChain(ctx, nextParentBlock, events)
 	if err != nil {
 		// NOTE: this is okay, it will auto-retry
 		return events, err
@@ -796,7 +796,10 @@ func (m *Monitor) fetchNextBlock(ctx context.Context) (*types.Block, []byte, boo
 
 	getter := func(ctx context.Context, _ string) ([]byte, error) {
 		if m.options.DebugLogging {
-			m.log.Debug(fmt.Sprintf("ethmonitor: fetchNextBlock is calling origin for number %s", m.nextBlockNumber))
+			m.nextBlockNumberMu.Lock()
+			nextNumForLog := m.nextBlockNumber
+			m.nextBlockNumberMu.Unlock()
+			m.log.Debug(fmt.Sprintf("ethmonitor: fetchNextBlock is calling origin for number %v", nextNumForLog))
 		}
 		for {
 			select {
@@ -805,9 +808,15 @@ func (m *Monitor) fetchNextBlock(ctx context.Context) (*types.Block, []byte, boo
 			default:
 			}
 
-			nextBlockPayload, err := m.fetchRawBlockByNumber(ctx, m.nextBlockNumber)
+			m.nextBlockNumberMu.Lock()
+			nextNum := m.nextBlockNumber
+			m.nextBlockNumberMu.Unlock()
+			nextBlockPayload, err := m.fetchRawBlockByNumber(ctx, nextNum)
 			if err != nil {
-				m.log.Debug(fmt.Sprintf("ethmonitor: [retrying] failed to fetch next block # %d, due to: %v", m.nextBlockNumber, err))
+				m.nextBlockNumberMu.Lock()
+				logNum := m.nextBlockNumber
+				m.nextBlockNumberMu.Unlock()
+				m.log.Debug(fmt.Sprintf("ethmonitor: [retrying] failed to fetch next block # %v, due to: %v", logNum, err))
 				miss = true
 				if m.IsStreamingMode() {
 					// in streaming mode, we'll use a shorter time to pause before we refetch
@@ -996,7 +1005,16 @@ func (m *Monitor) publish(ctx context.Context, events Blocks) error {
 	// Check for trail-behind-head mode and set maxBlockNum if applicable
 	maxBlockNum := uint64(0)
 	if m.options.TrailNumBlocksBehindHead > 0 {
-		maxBlockNum = m.LatestBlock().NumberU64() - uint64(m.options.TrailNumBlocksBehindHead)
+		latest := m.LatestBlock()
+		if latest != nil {
+			lb := latest.NumberU64()
+			trail := uint64(m.options.TrailNumBlocksBehindHead)
+			if lb > trail {
+				maxBlockNum = lb - trail
+			} else {
+				maxBlockNum = 0
+			}
+		}
 	}
 
 	// Enqueue

ethmonitor/logs_stream.go

@@ -0,0 +1,60 @@
+package ethmonitor
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"github.com/0xsequence/ethkit/ethrpc"
+	"github.com/0xsequence/ethkit/go-ethereum"
+	"github.com/0xsequence/ethkit/go-ethereum/core/types"
+)
+
+type LogsStream struct {
+	provider ethrpc.Interface
+}
+
+func NewLogsStream(provider ethrpc.Interface) *LogsStream {
+	return &LogsStream{
+		provider: provider,
+	}
+}
+
+// func (s *LogsStream) StreamLogs(ctx context.Context) (<-chan [][]types.Log, <-chan error) {
+// 	logsCh := make(chan [][]types.Log)
+// 	errCh := make(chan error, 1)
+//
+// 	go func() {
+// 		defer close(logsCh)
+// 		defer close(errCh)
+//
+// 	}()
+//
+// 	return logsCh, errCh
+// }
+
+func (s *LogsStream) Do(ctx context.Context) {
+	ch := make(chan types.Log)
+
+	sub, err := s.provider.SubscribeFilterLogs(ctx, ethereum.FilterQuery{}, ch)
+	if err != nil {
+		panic(err)
+	}
+	defer sub.Unsubscribe()
+
+	go func() {
+		for {
+			select {
+			case err := <-sub.Err():
+				panic(err)
+			case log := <-ch:
+				slog.Info("log received", "block", log.BlockNumber, "txIndex", log.TxIndex, "logIndex", log.Index)
+			}
+
+			// TODO: can a block tell us how many txns there are...?
+			// yes..
+		}
+	}()
+
+	time.Sleep(50 * time.Second)
+}