Insert in tx

cmd/decode.go

@@ -4,7 +4,7 @@ import (
 	"encoding/hex"
 	"fmt"
 
-	"github.com/lbryio/lbry.go/v2/schema/claim"
+	"github.com/lbryio/lbry.go/v2/schema/stake"
 
 	"github.com/davecgh/go-spew/spew"
 	"github.com/golang/protobuf/jsonpb"
@@ -23,7 +23,7 @@ func init() {
 }
 
 func decodeCmd(cmd *cobra.Command, args []string) {
-	c, err := claim.DecodeClaimHex(args[0], "")
+	c, err := stake.DecodeClaimHex(args[0], "")
 	if err != nil {
 		log.Fatal(err)
 	}

cmd/getstream.go

@@ -41,7 +41,7 @@ func getStreamCmd(cmd *cobra.Command, args []string) {
 
 	var sd stream.SDBlob
 
-	sdb, err := s.Get(sdHash)
+	sdb, _, err := s.Get(sdHash)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -62,7 +62,7 @@ func getStreamCmd(cmd *cobra.Command, args []string) {
 	}
 
 	for i := 0; i < len(sd.BlobInfos)-1; i++ {
-		b, err := s.Get(hex.EncodeToString(sd.BlobInfos[i].BlobHash))
+		b, _, err := s.Get(hex.EncodeToString(sd.BlobInfos[i].BlobHash))
 		if err != nil {
 			log.Fatal(err)
 		}

cmd/integrity.go

@@ -0,0 +1,93 @@
+package cmd
+
+import (
+	"crypto/sha512"
+	"encoding/hex"
+	"io/ioutil"
+	"os"
+	"path"
+	"runtime"
+	"sync/atomic"
+	"time"
+
+	"github.com/lbryio/lbry.go/v2/extras/errors"
+	"github.com/lbryio/reflector.go/meta"
+	"github.com/lbryio/reflector.go/store/speedwalk"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+)
+
+var threads int
+
+func init() {
+	var cmd = &cobra.Command{
+		Use:   "check-integrity",
+		Short: "check blobs integrity for a given path",
+		Run:   integrityCheckCmd,
+	}
+	cmd.Flags().StringVar(&diskStorePath, "store-path", "", "path of the store where all blobs are cached")
+	cmd.Flags().IntVar(&threads, "threads", runtime.NumCPU()-1, "number of concurrent threads to process blobs")
+	rootCmd.AddCommand(cmd)
+}
+
+func integrityCheckCmd(cmd *cobra.Command, args []string) {
+	log.Printf("reflector %s", meta.VersionString())
+	if diskStorePath == "" {
+		log.Fatal("store-path must be defined")
+	}
+
+	blobs, err := speedwalk.AllFiles(diskStorePath, true)
+	if err != nil {
+		log.Errorf("error while reading blobs from disk %s", errors.FullTrace(err))
+	}
+	tasks := make(chan string, len(blobs))
+	done := make(chan bool)
+	processed := new(int32)
+	go produce(tasks, blobs)
+	cpus := runtime.NumCPU()
+	for i := 0; i < cpus-1; i++ {
+		go consume(i, tasks, done, len(blobs), processed)
+	}
+	<-done
+}
+
+func produce(tasks chan<- string, blobs []string) {
+	for _, b := range blobs {
+		tasks <- b
+	}
+	close(tasks)
+}
+
+func consume(worker int, tasks <-chan string, done chan<- bool, totalTasks int, processed *int32) {
+	start := time.Now()
+
+	for b := range tasks {
+		checked := atomic.AddInt32(processed, 1)
+		if worker == 0 {
+			remaining := int32(totalTasks) - checked
+			timePerBlob := time.Since(start).Microseconds() / int64(checked)
+			remainingTime := time.Duration(int64(remaining)*timePerBlob) * time.Microsecond
+			log.Infof("[T%d] %d/%d blobs checked. ETA: %s", worker, checked, totalTasks, remainingTime.String())
+		}
+		blobPath := path.Join(diskStorePath, b[:2], b)
+		blob, err := ioutil.ReadFile(blobPath)
+		if err != nil {
+			if os.IsNotExist(err) {
+				continue
+			}
+			log.Errorf("[Worker %d] Error looking up blob %s: %s", worker, b, err.Error())
+			continue
+		}
+		hashBytes := sha512.Sum384(blob)
+		readHash := hex.EncodeToString(hashBytes[:])
+		if readHash != b {
+			log.Infof("[%s] found a broken blob while reading from disk. Actual hash: %s", b, readHash)
+			err := os.Remove(blobPath)
+			if err != nil {
+				log.Errorf("Error while deleting broken blob %s: %s", b, err.Error())
+			}
+		}
+	}
+	done <- true
+}

cmd/peer.go

@@ -33,11 +33,13 @@ func peerCmd(cmd *cobra.Command, args []string) {
 	peerServer := peer.NewServer(s3)
 
 	if !peerNoDB {
-		db := new(db.SQL)
+		db := &db.SQL{
+			LogQueries: log.GetLevel() == log.DebugLevel,
+		}
 		err = db.Connect(globalConfig.DBConn)
 		checkErr(err)
 
-		combo := store.NewDBBackedStore(s3, db)
+		combo := store.NewDBBackedStore(s3, db, false)
 		peerServer = peer.NewServer(combo)
 	}
 

cmd/populatedb.go

@@ -0,0 +1,47 @@
+package cmd
+
+import (
+	"github.com/lbryio/lbry.go/v2/extras/errors"
+	"github.com/lbryio/reflector.go/db"
+	"github.com/lbryio/reflector.go/meta"
+	"github.com/lbryio/reflector.go/store/speedwalk"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+)
+
+var (
+	diskStorePath string
+)
+
+func init() {
+	var cmd = &cobra.Command{
+		Use:   "populate-db",
+		Short: "populate local database with blobs from a disk storage",
+		Run:   populateDbCmd,
+	}
+	cmd.Flags().StringVar(&diskStorePath, "store-path", "",
+		"path of the store where all blobs are cached")
+	rootCmd.AddCommand(cmd)
+}
+
+func populateDbCmd(cmd *cobra.Command, args []string) {
+	log.Printf("reflector %s", meta.VersionString())
+	if diskStorePath == "" {
+		log.Fatal("store-path must be defined")
+	}
+	localDb := &db.SQL{
+		SoftDelete:  true,
+		TrackAccess: db.TrackAccessBlobs,
+		LogQueries:  log.GetLevel() == log.DebugLevel,
+	}
+	err := localDb.Connect("reflector:reflector@tcp(localhost:3306)/reflector")
+	if err != nil {
+		log.Fatal(err)
+	}
+	blobs, err := speedwalk.AllFiles(diskStorePath, true)
+	err = localDb.AddBlobs(blobs)
+	if err != nil {
+		log.Errorf("error while storing to db: %s", errors.FullTrace(err))
+	}
+}