prevent files from burning when rpc is down

jprov/server/file_server.go

@@ -240,20 +240,13 @@ func (f *FileServer) RecollectActiveDeals() error {
 		return err
 	}
 
-	count := 0
-
 	for _, q := range queryActiveDeals {
-		_, err := f.archivedb.GetFid(q.Cid)
-		if errors.Is(err, archive.ErrContractNotFound) {
-			err = f.archivedb.SetContract(q.Cid, q.Fid)
-			count++
-			if err != nil {
-				return err
-			}
+		err := f.archivedb.SetContract(q.Cid, q.Fid)
+		if err != nil && !errors.Is(err, archive.ErrContractAlreadyExists) {
+			return err
 		}
 	}
 
-	f.logger.Info(fmt.Sprintf("recollected deals: %d\n", count))
 	return nil
 }
 

jprov/server/proofs.go

@@ -378,19 +378,14 @@ func (f *FileServer) ContractState(cid string) string {
 	return f.QueryContractState(cid)
 }
 
-func (f *FileServer) Prove(cid string) error {
-	block, err := queryBlock(&f.cosmosCtx, string(cid))
-	if err != nil {
-		return err
-	}
-
-	dex, ok := sdk.NewIntFromString(block)
-	f.logger.Debug(fmt.Sprintf("BlockToProve: %s", block))
+func (f *FileServer) Prove(deal storageTypes.ActiveDeals) error {
+	dex, ok := sdk.NewIntFromString(deal.Blocktoprove)
+	f.logger.Debug(fmt.Sprintf("BlockToProve: %s", deal.Blocktoprove))
 	if !ok {
-		return fmt.Errorf("failed to parse block number: %s", block)
+		return fmt.Errorf("failed to parse block number: %s", deal.Blocktoprove)
 	}
 
-	return f.postProof(string(cid), f.blockSize, dex.Int64())
+	return f.postProof(deal.Cid, f.blockSize, dex.Int64())
 }
 
 func (f *FileServer) handleContracts() error {
@@ -405,31 +400,34 @@ func (f *FileServer) handleContracts() error {
 		}
 
 		f.logger.Info(fmt.Sprintf("CID: %s FID: %s", cid, fid))
+		resp, respErr := f.QueryActiveDeal(cid)
 
-		switch state := f.QueryContractState(cid); state {
-		case verified:
+		switch state, err := types.ContractState(resp, respErr); state {
+		case types.Verified:
 			err := f.DeleteDowntime(cid)
 			if err != nil {
-				f.logger.Error("error when unmarking downtime cid: ", cid, ": ", err)
+				f.logger.Error(fmt.Sprintf("error when unmarking downtime cid: %s: %v", cid, err))
 			}
 			continue
-		case notFound:
+		case types.NotFound:
 			err := f.IncrementDowntime(cid)
 			if err != nil {
 				return err
 			}
-		case notVerified:
+		case types.NotVerified:
 			err := f.DeleteDowntime(cid)
 			if err != nil {
-				f.logger.Error("error when unmarking downtime cid: ", cid, ": ", err)
+				f.logger.Error(fmt.Sprintf("error when unmarking downtime cid: %s: %v", cid, err))
 			}
 
-			err = f.Prove(cid)
+			err = f.Prove(resp.ActiveDeals)
 			if err != nil {
-				f.logger.Error("failed to prove ", cid, ": ", err)
+				f.logger.Error(fmt.Sprintf("failed to prove: %s: %v", cid, err))
 			}
+		case types.Error:
+			f.logger.Error(fmt.Sprintf("query error: %v", err))
 		default:
-			f.logger.Error("unknown state to handle: ", state)
+			return fmt.Errorf("unkown state: %v %v", state, err)
 		}
 	}
 	return nil
@@ -452,6 +450,7 @@ func (f *FileServer) StartProofServer(interval uint16) {
 	for {
 		select {
 		case <-sigChan:
+			fmt.Println("shutting down proof server")
 			return
 		default:
 			start := time.Now()

jprov/server/query.go

@@ -17,7 +17,6 @@ const (
 	notFound    = "not found"
 )
 
-// returns verified, notVerified or notFound
 func (f *FileServer) QueryContractState(cid string) string {
 	req := storageTypes.QueryActiveDealRequest{Cid: cid}
 	resp, err := f.queryClient.ActiveDeals(f.cmd.Context(), &req)
@@ -136,3 +135,8 @@ func (f *FileServer) QueryMyActiveDeals() ([]storageTypes.ActiveDeals, error) {
 
 	return filterMyActiveDeals(activeDeals, f.provider.Address), nil
 }
+
+func (f *FileServer) QueryActiveDeal(cid string) (*storageTypes.QueryActiveDealResponse, error) {
+	req := storageTypes.QueryActiveDealRequest{Cid: cid}
+	return f.queryClient.ActiveDeals(f.cmd.Context(), &req)
+}

jprov/server/utils.go

@@ -1,7 +1,6 @@
 package server
 
 import (
-	"context"
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
@@ -12,7 +11,6 @@ import (
 
 	"github.com/JackalLabs/jackal-provider/jprov/types"
 	"github.com/JackalLabs/jackal-provider/jprov/utils"
-	"github.com/cosmos/cosmos-sdk/client"
 	storageTypes "github.com/jackalLabs/canine-chain/v3/x/storage/types"
 )
 
@@ -78,23 +76,6 @@ func testConnection(providers []storageTypes.Providers, ip string) bool {
 	return true
 }
 
-func queryBlock(clientCtx *client.Context, cid string) (string, error) {
-	queryClient := storageTypes.NewQueryClient(clientCtx)
-
-	argCid := cid
-
-	params := &storageTypes.QueryActiveDealRequest{
-		Cid: argCid,
-	}
-
-	res, err := queryClient.ActiveDeals(context.Background(), params)
-	if err != nil {
-		return "", err
-	}
-
-	return res.ActiveDeals.Blocktoprove, nil
-}
-
 func buildCid(address, sender, fid string) (string, error) {
 	h := sha256.New()
 

jprov/types/state.go

@@ -0,0 +1,46 @@
+package types
+
+import (
+	"strconv"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	storageTypes "github.com/jackalLabs/canine-chain/v3/x/storage/types"
+)
+
+type VerifiedStatus int
+
+const (
+	Verified VerifiedStatus = iota
+	NotVerified
+	NotFound
+	Error
+)
+
+// Returns state of the active deal contract based on query response.
+// Returns Error status with non-nil error if respErr is unknown code or parsing resp failed.
+func ContractState(resp *storageTypes.QueryActiveDealResponse, respErr error) (VerifiedStatus, error) {
+	if respErr != nil {
+		stat, ok := status.FromError(respErr)
+		if !ok { // unknown grpc error
+			return Error, respErr
+		}
+
+		if codes.NotFound == stat.Code() {
+			return NotFound, nil
+		}
+
+		return Error, respErr
+	}
+
+	verified, err := strconv.ParseBool(resp.ActiveDeals.Proofverified)
+	if err != nil {
+		return Error, err
+	}
+
+	if verified {
+		return Verified, err
+	}
+	return NotVerified, err
+}