status-im/status-go

package rpcfilters

import (
    "fmt"
    "sort"
    "sync"

    "github.com/ethereum/go-ethereum/common"
    "github.com/ethereum/go-ethereum/core/types"
)

const (
    defaultCacheSize = 20
)

type cacheRecord struct {
    block uint64
    hash  common.Hash
    logs  []types.Log
}

func newCache(size int) *cache {
    return &cache{
        records: make([]cacheRecord, 0, size),
        size:    size,
    }
}

type cache struct {
    mu      sync.RWMutex
    size    int // length of the records
    records []cacheRecord
}

// add inserts logs into cache and returns added and replaced logs.
// replaced logs with will be returned with Removed=true.
func (c *cache) add(logs []types.Log) (added, replaced []types.Log, err error) {
    if len(logs) == 0 {
        return nil, nil, nil
    }
    aggregated := aggregateLogs(logs, c.size) // size doesn't change
    if len(aggregated) == 0 {
        return nil, nil, nil
    }
    if err := checkLogsAreInOrder(aggregated); err != nil {
        return nil, nil, err
    }
    c.mu.Lock()
    defer c.mu.Unlock()
    // find common block. e.g. [3,4] and [1,2,3,4] = 3
    last := 0
    if len(c.records) > 0 {
        last = len(c.records) - 1
        for aggregated[0].block < c.records[last].block && last > 0 {
            last--
        }
    }
    c.records, added, replaced = merge(last, c.records, aggregated)
    if lth := len(c.records); lth > c.size {
        copy(c.records, c.records[lth-c.size:])
    }
    return added, replaced, nil
}

func (c *cache) earliestBlockNum() uint64 {
    if len(c.records) == 0 {
        return 0
    }
    return c.records[0].block
}

func checkLogsAreInOrder(records []cacheRecord) error {
    for prev, i := 0, 1; i < len(records); i++ {
        if records[prev].block == records[i].block-1 {
            prev = i
        } else {
            return fmt.Errorf(
                "logs must be delivered straight in order. gaps between blocks '%d' and '%d'",
                records[prev].block, records[i].block,
            )
        }
    }
    return nil
}

// merge merges received records into old slice starting at provided position, example:
// [1, 2, 3]
//
//    [2, 3, 4]
//
// [1, 2, 3, 4]
// if hash doesn't match previously received hash - such block was removed due to reorg
// logs that were a part of that block will be returned with Removed set to true
func merge(last int, old, received []cacheRecord) ([]cacheRecord, []types.Log, []types.Log) {
    var (
        added, replaced []types.Log
        block           uint64
        hash            common.Hash
    )
    for i := range received {
        record := received[i]
        if last < len(old) {
            block = old[last].block
            hash = old[last].hash
        }
        if record.block > block {
            // simply add new records
            added = append(added, record.logs...)
            old = append(old, record)
        } else if record.hash != hash && record.block == block {
            // record hash is not equal to previous record hash at the same height
            // replace record in hash and add logs as replaced
            replaced = append(replaced, old[last].logs...)
            added = append(added, record.logs...)
            old[last] = record
        }
        last++
    }
    return old, added, replaced
}

// aggregateLogs creates at most requested amount of cacheRecords from provided logs.
// cacheRecords will be sorted in ascending order, starting from lowest block to highest.
func aggregateLogs(logs []types.Log, limit int) []cacheRecord {
    // sort in reverse order, so that iteration will start from latest blocks
    sort.Slice(logs, func(i, j int) bool {
        return logs[i].BlockNumber > logs[j].BlockNumber
    })
    rst := make([]cacheRecord, limit)
    pos, start := len(rst)-1, 0
    var hash common.Hash
    for i := range logs {
        log := logs[i]
        if (hash != common.Hash{}) && hash != log.BlockHash {
            rst[pos].logs = logs[start:i]
            start = i
            if pos-1 < 0 {
                break
            }
            pos--
        }
        rst[pos].logs = logs[start:]
        rst[pos].block = log.BlockNumber
        rst[pos].hash = log.BlockHash
        hash = log.BlockHash
    }
    return rst[pos:]
}