forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-23 13:16:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

eth/filters: almost working

Browse source 
Signed-off-by: jsvisa <delweng@gmail.com>
This commit is contained in:
jsvisa 2023-07-25 11:09:43 +08:00
parent c95bf8daf7
commit 3b3eb5aab5
1 changed files with 30 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

60
eth/filters/api.go
View file

@ -23,7 +23,6 @@ import (
"fmt" "fmt"
"math/big" "math/big"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum"
@ -337,9 +336,9 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
// 2. if a reorg occurs before the currently delivered block, then we need to stop the historical delivery, and send all replaced logs instead // 2. if a reorg occurs before the currently delivered block, then we need to stop the historical delivery, and send all replaced logs instead
var ( var (
liveLogs = make(chan []*types.Log) liveLogs = make(chan []*types.Log)
histLogs = make(chan []*types.Log) histLogs = make(chan *types.Log)
histErr = make(chan error) histDone = make(chan error)
reorged atomic.Bool closeC = make(chan struct{})
) )
liveLogsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), liveLogs) liveLogsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), liveLogs)
if err != nil { if err != nil {
@ -347,7 +346,7 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
} }
go func() { go func() {
histErr <- api.doHistLogs(from, crit, &reorged, histLogs) histDone <- api.doHistLogs(from, crit, histLogs, closeC)
}() }()
// Compose and notify the logs from liveLogs and histLogs // Compose and notify the logs from liveLogs and histLogs
@ -356,15 +355,17 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
delivered uint64 delivered uint64
reorgBlock uint64 reorgBlock uint64
liveOnly bool liveOnly bool
reorged bool
) )
for { for {
select { select {
case err := <-histErr: case err := <-histDone:
if err != nil { if err != nil {
logger.Warn("History logs delivery failed", "err", err)
liveLogsSub.Unsubscribe() liveLogsSub.Unsubscribe()
return return
} }
// else historical logs are all delivered, let's switch to live mode // Else historical logs are all delivered, let's switch to live mode
liveOnly = true liveOnly = true
histLogs = nil histLogs = nil
case logs := <-liveLogs: case logs := <-liveLogs:
@ -378,11 +379,15 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
} }
continue continue
} }
// TODO: if reorg occured in more than once?
if reorgBlock == 0 { if reorgBlock == 0 {
reorgBlock = logs[0].BlockNumber reorgBlock = logs[0].BlockNumber
reorged.Store(reorgBlock <= delivered) if reorgBlock <= delivered {
logger.Info("Reorg detected", "reorgBlock", reorgBlock, "delivered", delivered)
reorged = true
}
} }
if !reorged.Load() { if !reorged {
continue continue
} }
if logs[0].Removed { if logs[0].Removed {
@ -402,17 +407,20 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
notifier.Notify(rpcSub.ID, &log) notifier.Notify(rpcSub.ID, &log)
} }
} }
case logs := <-histLogs: case log := <-histLogs:
for _, log := range logs { if reorged {
log := log continue
delivered = log.BlockNumber
notifier.Notify(rpcSub.ID, &log)
} }
// TODO: deliver the same block logs
delivered = log.BlockNumber
notifier.Notify(rpcSub.ID, &log)
case <-rpcSub.Err(): // client send an unsubscribe request case <-rpcSub.Err(): // client send an unsubscribe request
liveLogsSub.Unsubscribe() liveLogsSub.Unsubscribe()
closeC <- struct{}{}
return return
case <-notifier.Closed(): // connection dropped case <-notifier.Closed(): // connection dropped
liveLogsSub.Unsubscribe() liveLogsSub.Unsubscribe()
closeC <- struct{}{}
return return
} }
} }
@ -421,12 +429,11 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
return nil return nil
} }
// doHistLogs retrieves the logs older than current header. // doHistLogs retrieves the logs older than current header, and forward them to the histLogs channel.
func (api *FilterAPI) doHistLogs(from int64, crit FilterCriteria, reorged *atomic.Bool, histLogs chan<- []*types.Log) error { func (api *FilterAPI) doHistLogs(from int64, crit FilterCriteria, histLogs chan<- *types.Log, closeC chan struct{}) error {
// The original request ctx will be canceled as soon as the parent goroutine // The original request ctx will be canceled as soon as the parent goroutine
// returns a subscription. Use a new context instead. // returns a subscription. Use a new context instead.
ctx, cancel := context.WithCancel(context.Background()) ctx := context.Background()
defer cancel()
for { for {
// Get the latest block header. // Get the latest block header.
@ -446,22 +453,15 @@ func (api *FilterAPI) doHistLogs(from int64, crit FilterCriteria, reorged *atomi
FORLOOP: FORLOOP:
for { for {
select { select {
case <-closeC:
return nil
case log := <-logsCh: case log := <-logsCh:
// TODO: deliver all logs of the current block when we detect a reorg histLogs <- log
if reorged.Load() {
logger.Debug("reorged, dropping log from old chain", "log", log)
cancel()
continue
}
histLogs <- []*types.Log{log}
case err := <-errChan: case err := <-errChan:
// Range filter is done or error, let's also stop the reorgLogs subscribe // Range filter is done or error, let's also stop the reorgLogs subscribe
if err != nil { if err != nil {
if err != context.Canceled { logger.Error("error while fetching historical logs", "err", err)
logger.Error("error while fetching historical logs", "err", err) return err
return err
}
return nil
} }
break FORLOOP break FORLOOP
} }