status-go/services/shhext/history.go

351 lines
10 KiB
Go

package shhext
import (
"errors"
"fmt"
"sort"
"sync"
"time"
"github.com/ethereum/go-ethereum/rlp"
"github.com/status-im/status-go/db"
"github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/mailserver"
)
const (
// WhisperTimeAllowance is needed to ensure that we won't miss envelopes that were
// delivered to mail server after we made a request.
WhisperTimeAllowance = 20 * time.Second
)
// NewHistoryUpdateReactor creates HistoryUpdateReactor instance.
func NewHistoryUpdateReactor() *HistoryUpdateReactor {
return &HistoryUpdateReactor{}
}
// HistoryUpdateReactor responsible for tracking progress for all history requests.
// It listens for 2 events:
// - when envelope from mail server is received we will update appropriate topic on disk
// - when confirmation for request completion is received - we will set last envelope timestamp as the last timestamp
// for all TopicLists in current request.
type HistoryUpdateReactor struct {
mu sync.Mutex
}
// UpdateFinishedRequest removes successfully finished request and updates every topic
// attached to the request.
func (reactor *HistoryUpdateReactor) UpdateFinishedRequest(ctx Context, id types.Hash) error {
reactor.mu.Lock()
defer reactor.mu.Unlock()
req, err := ctx.HistoryStore().GetRequest(id)
if err != nil {
return err
}
for i := range req.Histories() {
th := &req.Histories()[i]
th.RequestID = types.Hash{}
th.Current = th.End
th.End = time.Time{}
if err := th.Save(); err != nil {
return err
}
}
return req.Delete()
}
// UpdateTopicHistory updates Current timestamp for the TopicHistory with a given timestamp.
func (reactor *HistoryUpdateReactor) UpdateTopicHistory(ctx Context, topic types.TopicType, timestamp time.Time) error {
reactor.mu.Lock()
defer reactor.mu.Unlock()
histories, err := ctx.HistoryStore().GetHistoriesByTopic(topic)
if err != nil {
return err
}
if len(histories) == 0 {
return fmt.Errorf("no histories for topic 0x%x", topic)
}
for i := range histories {
th := &histories[i]
// this case could happen only iff envelopes were delivered out of order
// last envelope received, request completed, then others envelopes received
// request completed, last envelope received, and then all others envelopes received
if !th.Pending() {
continue
}
if timestamp.Before(th.End) && timestamp.After(th.Current) {
th.Current = timestamp
}
err := th.Save()
if err != nil {
return err
}
}
return nil
}
// TopicRequest defines what user has to provide.
type TopicRequest struct {
Topic types.TopicType
Duration time.Duration
}
// CreateRequests receives list of topic with desired timestamps and initiates both pending requests and requests
// that cover new topics.
func (reactor *HistoryUpdateReactor) CreateRequests(ctx Context, topicRequests []TopicRequest) ([]db.HistoryRequest, error) {
reactor.mu.Lock()
defer reactor.mu.Unlock()
seen := map[types.TopicType]struct{}{}
for i := range topicRequests {
if _, exist := seen[topicRequests[i].Topic]; exist {
return nil, errors.New("only one duration per topic is allowed")
}
seen[topicRequests[i].Topic] = struct{}{}
}
histories := map[types.TopicType]db.TopicHistory{}
for i := range topicRequests {
th, err := ctx.HistoryStore().GetHistory(topicRequests[i].Topic, topicRequests[i].Duration)
if err != nil {
return nil, err
}
histories[th.Topic] = th
}
requests, err := ctx.HistoryStore().GetAllRequests()
if err != nil {
return nil, err
}
filtered := []db.HistoryRequest{}
for i := range requests {
req := requests[i]
for _, th := range histories {
if th.Pending() {
delete(histories, th.Topic)
}
}
if !ctx.RequestRegistry().Has(req.ID) {
filtered = append(filtered, req)
}
}
adjusted, err := adjustRequestedHistories(ctx.HistoryStore(), mapToList(histories))
if err != nil {
return nil, err
}
filtered = append(filtered,
GroupHistoriesByRequestTimespan(ctx.HistoryStore(), adjusted)...)
return RenewRequests(filtered, ctx.Time()), nil
}
// for every history that is not included in any request check if there are other ranges with such topic in db
// if so check if they can be merged
// if not then adjust second part so that End of it will be equal to First of previous
func adjustRequestedHistories(store db.HistoryStore, histories []db.TopicHistory) ([]db.TopicHistory, error) {
adjusted := []db.TopicHistory{}
for i := range histories {
all, err := store.GetHistoriesByTopic(histories[i].Topic)
if err != nil {
return nil, err
}
th, err := adjustRequestedHistory(&histories[i], all...)
if err != nil {
return nil, err
}
if th != nil {
adjusted = append(adjusted, *th)
}
}
return adjusted, nil
}
func adjustRequestedHistory(th *db.TopicHistory, others ...db.TopicHistory) (*db.TopicHistory, error) {
sort.Slice(others, func(i, j int) bool {
return others[i].Duration > others[j].Duration
})
if len(others) == 1 && others[0].Duration == th.Duration {
return th, nil
}
for j := range others {
if others[j].Duration == th.Duration {
// skip instance with same duration
continue
} else if th.Duration > others[j].Duration {
if th.Current.Equal(others[j].First) {
// this condition will be reached when query for new index successfully finished
th.Current = others[j].Current
// FIXME next two db operations must be completed atomically
err := th.Save()
if err != nil {
return nil, err
}
err = others[j].Delete()
if err != nil {
return nil, err
}
} else if (others[j].First != time.Time{}) {
// select First timestamp with lowest value. if there are multiple indexes that cover such ranges:
// 6:00 - 7:00 Duration: 3h
// 7:00 - 8:00 2h
// 8:00 - 9:00 1h
// and client created new index with Duration 4h
// 4h index must have End value set to 6:00
if (others[j].First.Before(th.End) || th.End == time.Time{}) {
th.End = others[j].First
}
} else {
// remove previous if it is covered by new one
// client created multiple indexes without any succsefully executed query
err := others[j].Delete()
if err != nil {
return nil, err
}
}
} else if th.Duration < others[j].Duration {
if !others[j].Pending() {
th = &others[j]
} else {
return nil, nil
}
}
}
return th, nil
}
// RenewRequests re-sets current, first and end timestamps.
// Changes should not be persisted on disk in this method.
func RenewRequests(requests []db.HistoryRequest, now time.Time) []db.HistoryRequest {
zero := time.Time{}
for i := range requests {
req := requests[i]
histories := req.Histories()
for j := range histories {
history := &histories[j]
if history.Current == zero {
history.Current = now.Add(-(history.Duration))
}
if history.First == zero {
history.First = history.Current
}
if history.End == zero {
history.End = now
}
}
}
return requests
}
// CreateTopicOptionsFromRequest transforms histories attached to a single request to a simpler format - TopicOptions.
func CreateTopicOptionsFromRequest(req db.HistoryRequest) TopicOptions {
histories := req.Histories()
rst := make(TopicOptions, len(histories))
for i := range histories {
history := histories[i]
rst[i] = TopicOption{
Topic: history.Topic,
Range: Range{
Start: uint64(history.Current.Add(-(WhisperTimeAllowance)).Unix()),
End: uint64(history.End.Unix()),
},
}
}
return rst
}
func mapToList(topics map[types.TopicType]db.TopicHistory) []db.TopicHistory {
rst := make([]db.TopicHistory, 0, len(topics))
for key := range topics {
rst = append(rst, topics[key])
}
return rst
}
// GroupHistoriesByRequestTimespan creates requests from provided histories.
// Multiple histories will be included into the same request only if they share timespan.
func GroupHistoriesByRequestTimespan(store db.HistoryStore, histories []db.TopicHistory) []db.HistoryRequest {
requests := []db.HistoryRequest{}
for _, th := range histories {
var added bool
for i := range requests {
req := &requests[i]
histories := req.Histories()
if histories[0].SameRange(th) {
req.AddHistory(th)
added = true
}
}
if !added {
req := store.NewRequest()
req.AddHistory(th)
requests = append(requests, req)
}
}
return requests
}
// Range of the request.
type Range struct {
Start uint64
End uint64
}
// TopicOption request for a single topic.
type TopicOption struct {
Topic types.TopicType
Range Range
}
// TopicOptions is a list of topic-based requsts.
type TopicOptions []TopicOption
// ToBloomFilterOption creates bloom filter request from a list of topics.
func (options TopicOptions) ToBloomFilterOption() BloomFilterOption {
topics := make([]types.TopicType, len(options))
var start, end uint64
for i := range options {
opt := options[i]
topics[i] = opt.Topic
if opt.Range.Start > start {
start = opt.Range.Start
}
if opt.Range.End > end {
end = opt.Range.End
}
}
return BloomFilterOption{
Range: Range{Start: start, End: end},
Filter: topicsToBloom(topics...),
}
}
// Topics returns list of whisper TopicType attached to each TopicOption.
func (options TopicOptions) Topics() []types.TopicType {
rst := make([]types.TopicType, len(options))
for i := range options {
rst[i] = options[i].Topic
}
return rst
}
// BloomFilterOption is a request based on bloom filter.
type BloomFilterOption struct {
Range Range
Filter []byte
}
// ToMessagesRequestPayload creates mailserver.MessagesRequestPayload and encodes it to bytes using rlp.
func (filter BloomFilterOption) ToMessagesRequestPayload() ([]byte, error) {
// TODO fix this conversion.
// we start from time.Duration which is int64, then convert to uint64 for rlp-serilizability
// why uint32 here? max uint32 is smaller than max int64
payload := mailserver.MessagesRequestPayload{
Lower: uint32(filter.Range.Start),
Upper: uint32(filter.Range.End),
Bloom: filter.Filter,
// Client must tell the MailServer if it supports batch responses.
// This can be removed in the future.
Batch: true,
Limit: 1000,
}
return rlp.EncodeToBytes(payload)
}