package database import ( "context" "math/rand" "time" "github.com/ledgerwatch/erigon/cmd/observer/utils" "github.com/ledgerwatch/log/v3" ) type DBRetrier struct { db DB log log.Logger } func NewDBRetrier(db DB, logger log.Logger) DBRetrier { return DBRetrier{db, logger} } func retryBackoffTime(attempt int) time.Duration { if attempt <= 0 { return 0 } jitter := rand.Int63n(30 * time.Millisecond.Nanoseconds() * int64(attempt)) // nolint: gosec var ns int64 if attempt <= 6 { ns = ((50 * time.Millisecond.Nanoseconds()) << (attempt - 1)) + jitter } else { ns = 1600*time.Millisecond.Nanoseconds() + jitter } return time.Duration(ns) } func (db DBRetrier) retry(ctx context.Context, opName string, op func(context.Context) (interface{}, error)) (interface{}, error) { const retryCount = 40 return utils.Retry(ctx, retryCount, retryBackoffTime, db.db.IsConflictError, db.log, opName, op) } func (db DBRetrier) UpsertNodeAddr(ctx context.Context, id NodeID, addr NodeAddr) error { _, err := db.retry(ctx, "UpsertNodeAddr", func(ctx context.Context) (interface{}, error) { return nil, db.db.UpsertNodeAddr(ctx, id, addr) }) return err } func (db DBRetrier) FindNodeAddr(ctx context.Context, id NodeID) (*NodeAddr, error) { resultAny, err := db.retry(ctx, "FindNodeAddr", func(ctx context.Context) (interface{}, error) { return db.db.FindNodeAddr(ctx, id) }) if resultAny == nil { return nil, err } result := resultAny.(*NodeAddr) return result, err } func (db DBRetrier) ResetPingError(ctx context.Context, id NodeID) error { _, err := db.retry(ctx, "ResetPingError", func(ctx context.Context) (interface{}, error) { return nil, db.db.ResetPingError(ctx, id) }) return err } func (db DBRetrier) UpdatePingError(ctx context.Context, id NodeID) error { _, err := db.retry(ctx, "UpdatePingError", func(ctx context.Context) (interface{}, error) { return nil, db.db.UpdatePingError(ctx, id) }) return err } func (db DBRetrier) CountPingErrors(ctx context.Context, id NodeID) (*uint, error) { resultAny, err := db.retry(ctx, "CountPingErrors", func(ctx context.Context) (interface{}, error) { return db.db.CountPingErrors(ctx, id) }) if resultAny == nil { return nil, err } result := resultAny.(*uint) return result, err } func (db DBRetrier) UpdateClientID(ctx context.Context, id NodeID, clientID string) error { _, err := db.retry(ctx, "UpdateClientID", func(ctx context.Context) (interface{}, error) { return nil, db.db.UpdateClientID(ctx, id, clientID) }) return err } func (db DBRetrier) FindClientID(ctx context.Context, id NodeID) (*string, error) { resultAny, err := db.retry(ctx, "FindClientID", func(ctx context.Context) (interface{}, error) { return db.db.FindClientID(ctx, id) }) if resultAny == nil { return nil, err } result := resultAny.(*string) return result, err } func (db DBRetrier) UpdateNetworkID(ctx context.Context, id NodeID, networkID uint) error { _, err := db.retry(ctx, "UpdateNetworkID", func(ctx context.Context) (interface{}, error) { return nil, db.db.UpdateNetworkID(ctx, id, networkID) }) return err } func (db DBRetrier) UpdateEthVersion(ctx context.Context, id NodeID, ethVersion uint) error { _, err := db.retry(ctx, "UpdateEthVersion", func(ctx context.Context) (interface{}, error) { return nil, db.db.UpdateEthVersion(ctx, id, ethVersion) }) return err } func (db DBRetrier) UpdateHandshakeTransientError(ctx context.Context, id NodeID, hasTransientErr bool) error { _, err := db.retry(ctx, "UpdateHandshakeTransientError", func(ctx context.Context) (interface{}, error) { return nil, db.db.UpdateHandshakeTransientError(ctx, id, hasTransientErr) }) return err } func (db DBRetrier) InsertHandshakeError(ctx context.Context, id NodeID, handshakeErr string) error { _, err := db.retry(ctx, "InsertHandshakeError", func(ctx context.Context) (interface{}, error) { return nil, db.db.InsertHandshakeError(ctx, id, handshakeErr) }) return err } func (db DBRetrier) DeleteHandshakeErrors(ctx context.Context, id NodeID) error { _, err := db.retry(ctx, "DeleteHandshakeErrors", func(ctx context.Context) (interface{}, error) { return nil, db.db.DeleteHandshakeErrors(ctx, id) }) return err } func (db DBRetrier) FindHandshakeLastErrors(ctx context.Context, id NodeID, limit uint) ([]HandshakeError, error) { resultAny, err := db.retry(ctx, "FindHandshakeLastErrors", func(ctx context.Context) (interface{}, error) { return db.db.FindHandshakeLastErrors(ctx, id, limit) }) if resultAny == nil { return nil, err } result := resultAny.([]HandshakeError) return result, err } func (db DBRetrier) UpdateHandshakeRetryTime(ctx context.Context, id NodeID, retryTime time.Time) error { _, err := db.retry(ctx, "UpdateHandshakeRetryTime", func(ctx context.Context) (interface{}, error) { return nil, db.db.UpdateHandshakeRetryTime(ctx, id, retryTime) }) return err } func (db DBRetrier) FindHandshakeRetryTime(ctx context.Context, id NodeID) (*time.Time, error) { resultAny, err := db.retry(ctx, "FindHandshakeRetryTime", func(ctx context.Context) (interface{}, error) { return db.db.FindHandshakeRetryTime(ctx, id) }) if resultAny == nil { return nil, err } result := resultAny.(*time.Time) return result, err } func (db DBRetrier) CountHandshakeCandidates(ctx context.Context) (uint, error) { resultAny, err := db.retry(ctx, "CountHandshakeCandidates", func(ctx context.Context) (interface{}, error) { return db.db.CountHandshakeCandidates(ctx) }) if resultAny == nil { return 0, err } result := resultAny.(uint) return result, err } func (db DBRetrier) TakeHandshakeCandidates(ctx context.Context, limit uint) ([]NodeID, error) { resultAny, err := db.retry(ctx, "TakeHandshakeCandidates", func(ctx context.Context) (interface{}, error) { return db.db.TakeHandshakeCandidates(ctx, limit) }) if resultAny == nil { return nil, err } result := resultAny.([]NodeID) return result, err } func (db DBRetrier) UpdateForkCompatibility(ctx context.Context, id NodeID, isCompatFork bool) error { _, err := db.retry(ctx, "UpdateForkCompatibility", func(ctx context.Context) (interface{}, error) { return nil, db.db.UpdateForkCompatibility(ctx, id, isCompatFork) }) return err } func (db DBRetrier) UpdateNeighborBucketKeys(ctx context.Context, id NodeID, keys []string) error { _, err := db.retry(ctx, "UpdateNeighborBucketKeys", func(ctx context.Context) (interface{}, error) { return nil, db.db.UpdateNeighborBucketKeys(ctx, id, keys) }) return err } func (db DBRetrier) FindNeighborBucketKeys(ctx context.Context, id NodeID) ([]string, error) { resultAny, err := db.retry(ctx, "FindNeighborBucketKeys", func(ctx context.Context) (interface{}, error) { return db.db.FindNeighborBucketKeys(ctx, id) }) if resultAny == nil { return nil, err } result := resultAny.([]string) return result, err } func (db DBRetrier) UpdateSentryCandidatesLastEventTime(ctx context.Context, value time.Time) error { _, err := db.retry(ctx, "UpdateSentryCandidatesLastEventTime", func(ctx context.Context) (interface{}, error) { return nil, db.db.UpdateSentryCandidatesLastEventTime(ctx, value) }) return err } func (db DBRetrier) FindSentryCandidatesLastEventTime(ctx context.Context) (*time.Time, error) { resultAny, err := db.retry(ctx, "FindSentryCandidatesLastEventTime", func(ctx context.Context) (interface{}, error) { return db.db.FindSentryCandidatesLastEventTime(ctx) }) if resultAny == nil { return nil, err } result := resultAny.(*time.Time) return result, err } func (db DBRetrier) UpdateCrawlRetryTime(ctx context.Context, id NodeID, retryTime time.Time) error { _, err := db.retry(ctx, "UpdateCrawlRetryTime", func(ctx context.Context) (interface{}, error) { return nil, db.db.UpdateCrawlRetryTime(ctx, id, retryTime) }) return err } func (db DBRetrier) CountCandidates(ctx context.Context) (uint, error) { resultAny, err := db.retry(ctx, "CountCandidates", func(ctx context.Context) (interface{}, error) { return db.db.CountCandidates(ctx) }) if resultAny == nil { return 0, err } result := resultAny.(uint) return result, err } func (db DBRetrier) TakeCandidates(ctx context.Context, limit uint) ([]NodeID, error) { resultAny, err := db.retry(ctx, "TakeCandidates", func(ctx context.Context) (interface{}, error) { return db.db.TakeCandidates(ctx, limit) }) if resultAny == nil { return nil, err } result := resultAny.([]NodeID) return result, err } func (db DBRetrier) IsConflictError(err error) bool { return db.db.IsConflictError(err) }