Merge pull request #25 from rarimo/fix/ipfs-retry

Add retries for ipfs request

Author: 2pa-arch

config.yaml

@@ -40,8 +40,12 @@ voting_v2:
   gas_limit: 800000
   enable: true
   check_with_subscribe: false
-  ipfs_url: "https://ipfs.rarimo.com"
 
+ipfs:
+  url: "https://ipfs.rarimo.com/ipfs"
+  max_retries: 3
+  min_abnormal_period: 2s
+  max_abnormal_period: 8s
 
 replicator:
   address: 0x10f370A6d8782E0e0E85ba948be6DA2465Aab4E2

internal/config/ipfs.go

@@ -0,0 +1,43 @@
+package config
+
+import (
+	"time"
+
+	"gitlab.com/distributed_lab/figure/v3"
+	"gitlab.com/distributed_lab/kit/kv"
+	"gitlab.com/distributed_lab/logan/v3"
+	"gitlab.com/distributed_lab/logan/v3/errors"
+)
+
+const (
+	ipfsYamlKey = "ipfs"
+)
+
+type Ipfs struct {
+	Url               string        `fig:"url"`
+	MaxRetries        uint64        `fig:"max_retries"`
+	MinAbnormalPeriod time.Duration `fig:"min_abnormal_period"`
+	MaxAbnormalPeriod time.Duration `fig:"max_abnormal_period"`
+}
+
+func (c *config) Ipfs() Ipfs {
+	return c.ipfs.Do(func() interface{} {
+		result := Ipfs{
+			Url:               "https://ipfs.rarimo.com/ipfs",
+			MaxRetries:        5,
+			MinAbnormalPeriod: 30 * time.Second,
+			MaxAbnormalPeriod: 30 * time.Second,
+		}
+
+		err := figure.
+			Out(&result).
+			With(figure.BaseHooks).
+			From(kv.MustGetStringMap(c.getter, ipfsYamlKey)).
+			Please()
+		if err != nil {
+			panic(errors.Wrap(err, "failed to figure out config", logan.F{"key": ipfsYamlKey}))
+		}
+
+		return result
+	}).(Ipfs)
+}

internal/config/main.go

@@ -18,6 +18,7 @@ type Config interface {
 	VotingV2Configer
 	Pinger() Pinger
 	Replicator() Replicator
+	Ipfs() Ipfs
 }
 
 type config struct {
@@ -32,6 +33,7 @@ type config struct {
 
 	pinger     comfig.Once
 	replicator comfig.Once
+	ipfs       comfig.Once
 }
 
 func New(getter kv.Getter) Config {

internal/config/voting_v2_config.go

@@ -44,7 +44,6 @@ type VotingV2Config struct {
 	Block      uint64
 	Enable     bool
 	WithSub    bool
-	IpfsUrl    string
 }
 
 func (e *ethereumVoting) VotingV2Config() *VotingV2Config {
@@ -78,7 +77,6 @@ func (e *ethereumVoting) VotingV2Config() *VotingV2Config {
 			GasLimit       uint64            `fig:"gas_limit"`
 			Enable         bool              `fig:"enable"`
 			WithSub        bool              `fig:"check_with_subscribe"`
-			IpfsUrl        string            `fig:"ipfs_url"`
 		}{}
 		err := figure.
 			Out(&networkConfig).
@@ -110,7 +108,6 @@ func (e *ethereumVoting) VotingV2Config() *VotingV2Config {
 		result.mut = &sync.Mutex{}
 		result.Block = networkConfig.Block
 		result.GasLimit = networkConfig.GasLimit
-		result.IpfsUrl = networkConfig.IpfsUrl
 		return &result
 	}).(*VotingV2Config)
 }

internal/service/checker/process_logs.go

@@ -17,6 +17,8 @@ import (
 	"gitlab.com/distributed_lab/logan/v3/errors"
 )
 
+var DuplicateEventErr = errors.New("Duplicate event in db")
+
 // processLog based on the event name and filtered log, it parses and update the relevant information
 func (ch *checker) processLog(vLog types.Log, eventName string) error {
 	var processedEvent data.ProcessedEvent
@@ -27,6 +29,13 @@ func (ch *checker) processLog(vLog types.Log, eventName string) error {
 	processedEvent.TransactionHash = vLog.TxHash[:]
 	err := ch.insertProcessedEventLog(processedEvent)
 	if err != nil {
+		if err == DuplicateEventErr {
+			ch.log.WithFields(logan.F{
+				"hash_tx":   vLog.TxHash.Hex(),
+				"log_index": processedEvent.LogIndex,
+			}).Debug("event already processed and stored")
+			return nil
+		}
 		return errors.Wrap(err, "failed insert processed event")
 	}
 	sender, err := ch.getSender(vLog.TxHash)
@@ -149,7 +158,7 @@ func (ch *checker) getSender(txHash common.Hash) (string, error) {
 func (ch *checker) insertProcessedEventLog(processedEvent data.ProcessedEvent) error {
 	isExist, err := ch.processedEventQ.CheckProcessedEventExist(processedEvent)
 	if isExist {
-		return errors.New("Duplicate event in db")
+		return DuplicateEventErr
 	}
 	if err != nil {
 		return errors.Wrap(err, "failed to check processed event exist in db")

internal/service/checker/proposal_info.go

@@ -1,6 +1,7 @@
 package checker
 
 import (
+	"context"
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
@@ -16,6 +17,7 @@ import (
 	"github.com/rarimo/proof-verification-relayer/internal/data"
 	"github.com/rarimo/proof-verification-relayer/resources"
 	"gitlab.com/distributed_lab/logan/v3/errors"
+	"gitlab.com/distributed_lab/running"
 )
 
 type ProposalInfoFromContract struct {
@@ -34,7 +36,6 @@ type ProposalInfoFromContract struct {
 func GetProposalInfo(proposalId int64, cfg config.Config, creatorAddress string) (*ProposalInfoFromContract, error) {
 	contractAddress := cfg.VotingV2Config().Address
 	client := cfg.VotingV2Config().RPC
-	ipfsUrl := cfg.VotingV2Config().IpfsUrl
 	caller, err := contracts.NewProposalsStateCaller(contractAddress, client)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed get new ProposalsStateCaller")
@@ -49,9 +50,21 @@ func GetProposalInfo(proposalId int64, cfg config.Config, creatorAddress string)
 	endTimestamp := proposalInfo.Config.StartTimestamp + proposalInfo.Config.Duration
 
 	// Description in proposalInfo.Config contains the CID for the resource that corresponds to this proposal in the IPFS repository
-	desc, err := getProposalDescFromIpfs(proposalInfo.Config.Description, ipfsUrl)
-	if err != nil {
-		return nil, errors.Wrap(err, "failed get proposal info from ipfs")
+	proposalMetadata := &resources.ProposalInfoAttributesMetadata{}
+	var lastErr error
+	running.WithThreshold(context.Background(), cfg.Log(), "getProposalDescFromIpfs", func(ctx context.Context) (bool, error) {
+		desc, err := getProposalDescFromIpfs(proposalInfo.Config.Description, cfg.Ipfs().Url)
+		if err != nil {
+			lastErr = err
+			return false, errors.Wrap(err, "failed get proposal info from ipfs")
+		}
+
+		proposalMetadata = desc
+		return true, nil
+	}, cfg.Ipfs().MinAbnormalPeriod, cfg.Ipfs().MaxAbnormalPeriod, cfg.Ipfs().MaxRetries)
+
+	if lastErr != nil {
+		return nil, errors.Wrap(lastErr, "failed get proposal info from ipfs")
 	}
 	votingConfig := proposalInfo.Config
 
@@ -118,7 +131,7 @@ func GetProposalInfo(proposalId int64, cfg config.Config, creatorAddress string)
 	return &ProposalInfoFromContract{
 		StartTimestamp:            int64(startTimeStamp),
 		EndTimestamp:              int64(endTimestamp),
-		Metadata:                  *desc,
+		Metadata:                  *proposalMetadata,
 		ParsedVotingWhitelistData: parsedVotingWhitelistData,
 	}, nil
 }
@@ -228,22 +241,26 @@ func getBoundaryInYears(boundaryDateHex string, startTimeStamp uint64) (int64, e
 
 // getProposalDescFromIpfs requests and returns proposal information from ipfs by CID
 func getProposalDescFromIpfs(desId string, ipfsUrl string) (*resources.ProposalInfoAttributesMetadata, error) {
+	var data resources.ProposalInfoAttributesMetadata
 	requestURL := fmt.Sprintf("%s/%s", ipfsUrl, desId)
+
 	resp, err := http.Get(requestURL)
 	if err != nil {
-		return nil, fmt.Errorf("error making http request: %v", err)
+		return nil, fmt.Errorf("failed to make request: %v", err)
 	}
 
-	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("non-200 response from ipfs: %d", resp.StatusCode)
+	}
 
 	body, err := io.ReadAll(resp.Body)
+	resp.Body.Close()
 	if err != nil {
-		return nil, fmt.Errorf("failed read body response: %v", err)
+		return nil, fmt.Errorf("failed to read body: %v", err)
 	}
 
-	var data resources.ProposalInfoAttributesMetadata
 	if err := json.Unmarshal(body, &data); err != nil {
-		return nil, fmt.Errorf("failed parse JSON: %v", err)
+		return nil, fmt.Errorf("failed to parse JSON: %v", err)
 	}
 
 	return &data, nil