update dependencies

update dependencies

.github/workflows/go.yml

@@ -0,0 +1,37 @@
+name: Go
+
+on:
+  push:
+    branches: [ master ]
+  pull_request:
+    branches: [ master ]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+
+      - name: Set up Go
+        uses: actions/setup-go@v3
+        with:
+          go-version: 1.20.x
+
+      - name: Build linux
+        run: make linux
+
+      - name: Build macos
+        run: make macos
+
+      - name: Test
+        run: make test
+
+      - name: Lint
+        run: make lint
+
+      - name: retrieve all tags
+        run: git fetch --prune --unshallow --tags
+
+      - name: Print changes since last version
+        run: git log $(git describe --tags --abbrev=0)..HEAD --no-merges --oneline

.github/workflows/release.yml

@@ -0,0 +1,62 @@
+name: release
+
+on:
+  push:
+    tags:
+      - "*.*.*"
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+
+      - name: Set up Go
+        uses: actions/setup-go@v3
+        with:
+          go-version: 1.20.x
+
+      - name: Build linux
+        run: make linux
+
+      - name: Build macos
+        run: make macos
+
+      - name: Test
+        run: make test
+
+      - name: Lint
+        run: make lint
+
+      - name: Zip macos
+        run: zip -r reflector_darwin_amd64.zip ./dist/darwin_amd64
+
+      - name: Zip linux
+        run: zip -r reflector_linux_amd64.zip ./dist/linux_amd64
+
+      - name: retrieve all tags
+        run: git fetch --prune --unshallow --tags
+
+      - name: Generate Changelog
+        run: git log $(git describe --tags --abbrev=0 @^)..@ --no-merges --oneline > ${{ github.workspace }}-CHANGELOG.txt
+
+      - name: upload to github releases
+        uses: softprops/action-gh-release@v1
+        with:
+          files: |
+            ./reflector_linux_amd64.zip
+            ./reflector_darwin_amd64.zip
+          body_path: ${{ github.workspace }}-CHANGELOG.txt
+
+#      - name: Login to DockerHub
+#        uses: docker/login-action@v2
+#        with:
+#          username: ${{ secrets.DOCKERHUB_USERNAME }}
+#          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
+#      - name: Generate docker image
+#        run: make image
+
+#      - name: Docker push
+#        run: make publish_image

.gitignore

@@ -1,3 +1,4 @@
 /vendor
 /config.json*
+/dist
 /bin

.travis.yml

@@ -1,12 +1,9 @@
 os: linux
-dist: trusty
+dist: bionic
 language: go
 
-env:
-  - GO111MODULE=on
-
 go:
-  - 1.15.x
+  - 1.20.x
 
 cache:
   directories:
@@ -17,7 +14,7 @@ notifications:
   email: false
 
 # Skip the install step. Don't `go get` dependencies. Only build with the code in vendor/
-install: true
+#install: true
 
 # Anything in before_script that returns a nonzero exit code will
 # flunk the build and immediately stop. It's sorta like having
@@ -25,14 +22,14 @@ install: true
 before_script:
 # All the .go files, excluding vendor/ and model (auto generated)
   - GO_FILES=$(find . -iname '*.go' ! -iname '*_test.go' -type f | grep -v /vendor/ ) #i wish we were this crazy :p
-  - go get golang.org/x/tools/cmd/goimports                     # Used in build script for generated files
+  - go install golang.org/x/tools/cmd/goimports                     # Used in build script for generated files
 # - go get github.com/golang/lint/golint                        # Linter
 #  - go get honnef.co/go/tools/cmd/megacheck                     # Badass static analyzer/linter
-  - go get github.com/jgautheron/gocyclo                        # Check against high complexity
-  - go get github.com/mdempsky/unconvert                        # Identifies unnecessary type conversions
-  - go get github.com/kisielk/errcheck                          # Checks for unhandled errors
-  - go get github.com/opennota/check/cmd/varcheck               # Checks for unused vars
-  - go get github.com/opennota/check/cmd/structcheck            # Checks for unused fields in structs
+  - go install github.com/fzipp/gocyclo/cmd/gocyclo@latest                 # Check against high complexity
+  - go install github.com/mdempsky/unconvert@latest                        # Identifies unnecessary type conversions
+  - go install github.com/kisielk/errcheck@latest                          # Checks for unhandled errors
+  - go install github.com/opennota/check/cmd/varcheck@latest               # Checks for unused vars
+  - go install github.com/opennota/check/cmd/structcheck@latest            # Checks for unused fields in structs
 
 
 
@@ -40,7 +37,7 @@ before_script:
 # in a modern Go project.
 script:
   # Fail if a .go file hasn't been formatted with gofmt
-  - test -z $(gofmt -s -l $GO_FILES)
+  - for i in $GO_FILES; do test -z $(gofmt -s -l $i); done
   # Run unit tests
   - make test
   # Checks for unused vars and fields on structs
@@ -59,11 +56,11 @@ script:
   # one last linter - ignore autogen code
   #- golint -set_exit_status $(go list ./... | grep -v /vendor/ )
   # Finally, build the binary
-  - make
+  - make linux
 
 deploy:
   - provider: s3
-    local_dir: ./bin
+    local_dir: ./dist/linux_amd64
     skip_cleanup: true
     on:
       repo: lbryio/reflector.go

Dockerfile

@@ -3,7 +3,7 @@ EXPOSE 8080
 
 RUN mkdir /app
 WORKDIR /app
-COPY bin/prism-bin ./prism
+COPY dist/linux_amd64/prism-bin ./prism
 RUN chmod +x prism
 
 ENTRYPOINT [ "/app/prism" ]

Makefile

@@ -1,25 +1,33 @@
+version := $(shell git describe --dirty --always --long --abbrev=7)
+commit := $(shell git rev-parse --short HEAD)
+commit_long := $(shell git rev-parse HEAD)
+branch := $(shell git rev-parse --abbrev-ref HEAD)
+curTime := $(shell date +%s)
+
 BINARY=prism-bin
-
-DIR = $(shell cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd)
-BIN_DIR = ${DIR}/bin
 IMPORT_PATH = github.com/lbryio/reflector.go
+LDFLAGS="-X ${IMPORT_PATH}/meta.version=$(version) -X ${IMPORT_PATH}/meta.commit=$(commit) -X ${IMPORT_PATH}/meta.commitLong=$(commit_long) -X ${IMPORT_PATH}/meta.branch=$(branch) -X '${IMPORT_PATH}/meta.Time=$(curTime)'"
+DIR = $(shell cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd)
+BIN_DIR = $(DIR)/dist
 
-VERSION = $(shell git --git-dir=${DIR}/.git describe --dirty --always --long --abbrev=7)
-LDFLAGS = -ldflags "-X ${IMPORT_PATH}/meta.Version=${VERSION} -X ${IMPORT_PATH}/meta.Time=$(shell date +%s)"
-
-
-.PHONY: build clean test lint
-.DEFAULT_GOAL: build
-
-
-build:
-	mkdir -p ${BIN_DIR} && CGO_ENABLED=0 go build ${LDFLAGS} -asmflags -trimpath=${DIR} -o ${BIN_DIR}/${BINARY} main.go
-
-clean:
-	if [ -f ${BIN_DIR}/${BINARY} ]; then rm ${BIN_DIR}/${BINARY}; fi
+.DEFAULT_GOAL := linux
 
+.PHONY: test
 test:
-	go test ./... -v -cover
+	go test -cover -v ./...
 
+.PHONY: lint
 lint:
-	go get github.com/alecthomas/gometalinter && gometalinter --install && gometalinter ./...
+	./scripts/lint.sh
+
+.PHONY: linux
+linux:
+	GOARCH=amd64 GOOS=linux go build -ldflags ${LDFLAGS} -asmflags -trimpath=${DIR} -o ${BIN_DIR}/linux_amd64/${BINARY}
+
+.PHONY: macos
+macos:
+	GOARCH=amd64 GOOS=darwin go build -ldflags ${LDFLAGS} -asmflags -trimpath=${DIR} -o ${BIN_DIR}/darwin_amd64/${BINARY}
+
+.PHONY: image
+image:
+	docker buildx build -t lbry/reflector:$(version) -t lbry/reflector:latest --platform linux/amd64 .

cluster/cluster.go

@@ -1,7 +1,7 @@
 package cluster
 
 import (
-	"io/ioutil"
+	"io"
 	baselog "log"
 	"sort"
 	"time"
@@ -52,7 +52,7 @@ func (c *Cluster) Connect() error {
 	conf.MemberlistConfig.AdvertisePort = c.port
 	conf.NodeName = c.name
 
-	nullLogger := baselog.New(ioutil.Discard, "", 0)
+	nullLogger := baselog.New(io.Discard, "", 0)
 	conf.Logger = nullLogger
 
 	c.eventCh = make(chan serf.Event)

cmd/integrity.go

@@ -3,7 +3,6 @@ package cmd
 import (
 	"crypto/sha512"
 	"encoding/hex"
-	"io/ioutil"
 	"os"
 	"path"
 	"runtime"
@@ -72,7 +71,7 @@ func consume(worker int, tasks <-chan string, done chan<- bool, totalTasks int,
 			log.Infof("[T%d] %d/%d blobs processed so far. ETA: %s", worker, processedSoFar, totalTasks, remainingTime.String())
 		}
 		blobPath := path.Join(diskStorePath, b[:2], b)
-		blob, err := ioutil.ReadFile(blobPath)
+		blob, err := os.ReadFile(blobPath)
 		if err != nil {
 			if os.IsNotExist(err) {
 				continue

cmd/reflector.go

@@ -14,7 +14,7 @@ import (
 	"github.com/lbryio/reflector.go/meta"
 	"github.com/lbryio/reflector.go/reflector"
 	"github.com/lbryio/reflector.go/server/http"
-	http32 "github.com/lbryio/reflector.go/server/http3"
+	"github.com/lbryio/reflector.go/server/http3"
 	"github.com/lbryio/reflector.go/server/peer"
 	"github.com/lbryio/reflector.go/store"
 
@@ -43,6 +43,7 @@ var (
 	//upstream configuration
 	upstreamReflector string
 	upstreamProtocol  string
+	upstreamEdgeToken string
 
 	//downstream configuration
 	requestQueueSize int
@@ -84,6 +85,7 @@ func init() {
 
 	cmd.Flags().StringVar(&upstreamReflector, "upstream-reflector", "", "host:port of a reflector server where blobs are fetched from")
 	cmd.Flags().StringVar(&upstreamProtocol, "upstream-protocol", "http", "protocol used to fetch blobs from another upstream reflector server (tcp/http3/http)")
+	cmd.Flags().StringVar(&upstreamEdgeToken, "upstream-edge-token", "", "token used to retrieve/authenticate protected content")
 
 	cmd.Flags().IntVar(&requestQueueSize, "request-queue-size", 200, "How many concurrent requests from downstream should be handled at once (the rest will wait)")
 
@@ -123,14 +125,14 @@ func reflectorCmd(cmd *cobra.Command, args []string) {
 	}
 	defer peerServer.Shutdown()
 
-	http3PeerServer := http32.NewServer(underlyingStoreWithCaches, requestQueueSize)
+	http3PeerServer := http3.NewServer(underlyingStoreWithCaches, requestQueueSize)
 	err = http3PeerServer.Start(":" + strconv.Itoa(http3PeerPort))
 	if err != nil {
 		log.Fatal(err)
 	}
 	defer http3PeerServer.Shutdown()
 
-	httpServer := http.NewServer(underlyingStoreWithCaches, requestQueueSize)
+	httpServer := http.NewServer(store.WithSingleFlight("sf-http", underlyingStoreWithCaches), requestQueueSize, upstreamEdgeToken)
 	err = httpServer.Start(":" + strconv.Itoa(httpPeerPort))
 	if err != nil {
 		log.Fatal(err)
@@ -162,12 +164,12 @@ func initUpstreamStore() store.BlobStore {
 			Timeout: 30 * time.Second,
 		})
 	case "http3":
-		s = http32.NewStore(http32.StoreOpts{
+		s = http3.NewStore(http3.StoreOpts{
 			Address: upstreamReflector,
 			Timeout: 30 * time.Second,
 		})
 	case "http":
-		s = store.NewHttpStore(upstreamReflector)
+		s = store.NewHttpStore(upstreamReflector, upstreamEdgeToken)
 	default:
 		log.Fatalf("protocol is not recognized: %s", upstreamProtocol)
 	}

cmd/root.go

@@ -2,7 +2,6 @@ package cmd
 
 import (
 	"encoding/json"
-	"io/ioutil"
 	"os"
 	"strings"
 
@@ -25,6 +24,7 @@ type Config struct {
 	BucketName   string `json:"bucket_name"`
 	DBConn       string `json:"db_conn"`
 	SlackHookURL string `json:"slack_hook_url"`
+	SlackChannel string `json:"slack_channel"`
 	UpdateBinURL string `json:"update_bin_url"`
 	UpdateCmd    string `json:"update_cmd"`
 }
@@ -102,7 +102,7 @@ func preRun(cmd *cobra.Command, args []string) {
 		hook := &slackrus.SlackrusHook{
 			HookURL:        globalConfig.SlackHookURL,
 			AcceptedLevels: slackrus.LevelThreshold(logrus.InfoLevel),
-			Channel:        "#reflector-logs",
+			Channel:        globalConfig.SlackChannel,
 			//IconEmoji:      ":ghost:",
 			//Username:       "reflector.go",
 		}
@@ -141,7 +141,7 @@ func argFuncs(funcs ...cobra.PositionalArgs) cobra.PositionalArgs {
 func loadConfig(path string) (Config, error) {
 	var c Config
 
-	raw, err := ioutil.ReadFile(path)
+	raw, err := os.ReadFile(path)
 	if err != nil {
 		if os.IsNotExist(err) {
 			return c, errors.Err("config file not found")

cmd/send.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"os"
 	"os/signal"
 	"path"
@@ -39,13 +38,12 @@ func sendCmd(cmd *cobra.Command, args []string) {
 	reflectorAddress := args[0]
 	err := hackyReflector.Connect(reflectorAddress)
 	checkErr(err)
-	defer hackyReflector.Close()
+	defer func() { _ = hackyReflector.Close() }()
 
 	filePath := args[1]
 	file, err := os.Open(filePath)
 	checkErr(err)
-	defer file.Close()
-
+	defer func() { _ = file.Close() }()
 	sdCachePath := ""
 	sdCacheDir := mustGetFlagString(cmd, "sd-cache")
 	if sdCacheDir != "" {
@@ -60,7 +58,7 @@ func sendCmd(cmd *cobra.Command, args []string) {
 
 	if sdCachePath != "" {
 		if _, err := os.Stat(sdCachePath); !os.IsNotExist(err) {
-			sdBlob, err := ioutil.ReadFile(sdCachePath)
+			sdBlob, err := os.ReadFile(sdCachePath)
 			checkErr(err)
 			cachedSDBlob := &stream.SDBlob{}
 			err = cachedSDBlob.FromBlob(sdBlob)
@@ -110,7 +108,7 @@ func sendCmd(cmd *cobra.Command, args []string) {
 	sd := enc.SDBlob()
 	//sd.StreamName = filepath.Base(filePath)
 	//sd.SuggestedFileName = filepath.Base(filePath)
-	err = ioutil.WriteFile(sdCachePath, sd.ToBlob(), 0666)
+	err = os.WriteFile(sdCachePath, sd.ToBlob(), 0666)
 	if err != nil {
 		fmt.Printf("error saving sd blob: %v\n", err)
 		fmt.Println(sd.ToJson())

cmd/sendblob.go

@@ -51,7 +51,7 @@ func sendBlobCmd(cmd *cobra.Command, args []string) {
 
 	file, err := os.Open(path)
 	checkErr(err)
-	defer file.Close()
+	defer func() { _ = file.Close() }()
 	s, err := stream.New(file)
 	checkErr(err)
 

cmd/version.go

@@ -18,5 +18,5 @@ func init() {
 }
 
 func versionCmd(cmd *cobra.Command, args []string) {
-	fmt.Println(meta.VersionString())
+	fmt.Println(meta.FullName())
 }

db/db.go

@@ -17,7 +17,7 @@ import (
 	"github.com/go-sql-driver/mysql"
 	_ "github.com/go-sql-driver/mysql" // blank import for db driver ensures its imported even if its not used
 	log "github.com/sirupsen/logrus"
-	"github.com/volatiletech/null"
+	"github.com/volatiletech/null/v8"
 	"go.uber.org/atomic"
 )
 

go.mod

@@ -1,51 +1,124 @@
 module github.com/lbryio/reflector.go
 
+go 1.20
+
 replace github.com/btcsuite/btcd => github.com/lbryio/lbrycrd.go v0.0.0-20200203050410-e1076f12bf19
 
 require (
-	github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878 // indirect
-	github.com/aws/aws-sdk-go v1.27.0
+	github.com/aws/aws-sdk-go v1.45.24
 	github.com/bluele/gcache v0.0.2
 	github.com/brk0v/directio v0.0.0-20190225130936-69406e757cf7
 	github.com/btcsuite/btcd v0.0.0-20190824003749-130ea5bddde3
 	github.com/btcsuite/btcutil v0.0.0-20190425235716-9e5f4b9a998d
-	github.com/c2h5oh/datasize v0.0.0-20200825124411-48ed595a09d2
+	github.com/c2h5oh/datasize v0.0.0-20220606134207-859f65c6625b
 	github.com/davecgh/go-spew v1.1.1
 	github.com/ekyoung/gin-nice-recovery v0.0.0-20160510022553-1654dca486db
-	github.com/gin-gonic/gin v1.7.1
-	github.com/go-sql-driver/mysql v1.6.0
-	github.com/gogo/protobuf v1.2.1
-	github.com/golang/protobuf v1.5.2
-	github.com/google/go-cmp v0.5.6 // indirect
-	github.com/google/gops v0.3.18
+	github.com/gin-gonic/gin v1.9.1
+	github.com/go-sql-driver/mysql v1.7.1
+	github.com/gogo/protobuf v1.3.2
+	github.com/golang/protobuf v1.5.3
+	github.com/google/gops v0.3.28
 	github.com/gorilla/mux v1.8.0
-	github.com/hashicorp/go-msgpack v0.5.5 // indirect
-	github.com/hashicorp/golang-lru v0.5.4 // indirect
-	github.com/hashicorp/serf v0.9.5
+	github.com/hashicorp/serf v0.10.1
 	github.com/inconshreveable/go-update v0.0.0-20160112193335-8152e7eb6ccf
-	github.com/johntdyer/slackrus v0.0.0-20210521205746-42486fb4c48c
-	github.com/karrick/godirwalk v1.16.1
-	github.com/lbryio/chainquery v1.9.0
-	github.com/lbryio/lbry.go/v2 v2.7.2-0.20210416195322-6516df1418e3
-	github.com/lbryio/types v0.0.0-20201019032447-f0b4476ef386
-	github.com/lucas-clemente/quic-go v0.20.1
-	github.com/phayes/freeport v0.0.0-20171002185219-e27662a4a9d6
-	github.com/prometheus/client_golang v1.10.0
-	github.com/sergi/go-diff v1.2.0 // indirect
-	github.com/sirupsen/logrus v1.8.1
-	github.com/spf13/afero v1.4.1 // indirect
-	github.com/spf13/cast v1.3.0
-	github.com/spf13/cobra v1.1.3
-	github.com/spf13/viper v1.7.1 // indirect
-	github.com/stretchr/testify v1.7.0
-	github.com/volatiletech/null v8.0.0+incompatible
-	go.uber.org/atomic v1.7.0
-	golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 // indirect
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
-	golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c // indirect
-	golang.org/x/text v0.3.6 // indirect
-	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
-	google.golang.org/protobuf v1.27.1 // indirect
+	github.com/johntdyer/slackrus v0.0.0-20230315191314-80bc92dee4fc
+	github.com/karrick/godirwalk v1.17.0
+	github.com/lbryio/chainquery v1.9.1-0.20230515181855-2fcba3115cfe
+	github.com/lbryio/lbry.go/v2 v2.7.2-0.20230307181431-a01aa6dc0629
+	github.com/lbryio/types v0.0.0-20220224142228-73610f6654a6
+	github.com/phayes/freeport v0.0.0-20220201140144-74d24b5ae9f5
+	github.com/prometheus/client_golang v1.16.0
+	github.com/quic-go/quic-go v0.39.0
+	github.com/sirupsen/logrus v1.9.3
+	github.com/spf13/cast v1.5.1
+	github.com/spf13/cobra v1.7.0
+	github.com/stretchr/testify v1.8.4
+	github.com/volatiletech/null/v8 v8.1.2
+	go.uber.org/atomic v1.11.0
+	golang.org/x/sync v0.4.0
 )
 
-go 1.16
+require (
+	github.com/armon/go-metrics v0.4.0 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f // indirect
+	github.com/btcsuite/go-socks v0.0.0-20170105172521-4720035b7bfd // indirect
+	github.com/btcsuite/websocket v0.0.0-20150119174127-31079b680792 // indirect
+	github.com/bytedance/sonic v1.9.1 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect
+	github.com/friendsofgo/errors v0.9.2 // indirect
+	github.com/fsnotify/fsnotify v1.6.0 // indirect
+	github.com/gabriel-vasile/mimetype v1.4.2 // indirect
+	github.com/gin-contrib/sse v0.1.0 // indirect
+	github.com/go-errors/errors v1.4.2 // indirect
+	github.com/go-ini/ini v1.67.0 // indirect
+	github.com/go-playground/locales v0.14.1 // indirect
+	github.com/go-playground/universal-translator v0.18.1 // indirect
+	github.com/go-playground/validator/v10 v10.14.0 // indirect
+	github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 // indirect
+	github.com/goccy/go-json v0.10.2 // indirect
+	github.com/gofrs/uuid v4.2.0+incompatible // indirect
+	github.com/google/btree v1.0.0 // indirect
+	github.com/google/pprof v0.0.0-20211214055906-6f57359322fd // indirect
+	github.com/gorilla/rpc v1.2.0 // indirect
+	github.com/gorilla/websocket v1.4.2 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
+	github.com/hashicorp/go-msgpack v0.5.3 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/hashicorp/go-sockaddr v1.0.0 // indirect
+	github.com/hashicorp/golang-lru v0.5.4 // indirect
+	github.com/hashicorp/hcl v1.0.0 // indirect
+	github.com/hashicorp/memberlist v0.5.0 // indirect
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
+	github.com/johntdyer/slack-go v0.0.0-20230314151037-c5bf334f9b6e // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.4 // indirect
+	github.com/leodido/go-urn v1.2.4 // indirect
+	github.com/lyoshenka/bencode v0.0.0-20180323155644-b7abd7672df5 // indirect
+	github.com/magiconair/properties v1.8.7 // indirect
+	github.com/mattn/go-isatty v0.0.19 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
+	github.com/miekg/dns v1.1.41 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/onsi/ginkgo/v2 v2.9.5 // indirect
+	github.com/pelletier/go-toml/v2 v2.0.8 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_model v0.3.0 // indirect
+	github.com/prometheus/common v0.42.0 // indirect
+	github.com/prometheus/procfs v0.10.1 // indirect
+	github.com/quic-go/qpack v0.4.0 // indirect
+	github.com/quic-go/qtls-go1-20 v0.3.4 // indirect
+	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
+	github.com/shopspring/decimal v1.3.1 // indirect
+	github.com/slack-go/slack v0.12.1 // indirect
+	github.com/spf13/afero v1.9.3 // indirect
+	github.com/spf13/jwalterweatherman v1.1.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/spf13/viper v1.15.0 // indirect
+	github.com/subosito/gotenv v1.4.2 // indirect
+	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
+	github.com/ugorji/go/codec v1.2.11 // indirect
+	github.com/volatiletech/inflect v0.0.1 // indirect
+	github.com/volatiletech/randomize v0.0.1 // indirect
+	github.com/volatiletech/strmangle v0.0.4 // indirect
+	go.uber.org/mock v0.3.0 // indirect
+	golang.org/x/arch v0.3.0 // indirect
+	golang.org/x/crypto v0.9.0 // indirect
+	golang.org/x/exp v0.0.0-20221205204356-47842c84f3db // indirect
+	golang.org/x/mod v0.11.0 // indirect
+	golang.org/x/net v0.10.0 // indirect
+	golang.org/x/sys v0.11.0 // indirect
+	golang.org/x/text v0.9.0 // indirect
+	golang.org/x/time v0.3.0 // indirect
+	golang.org/x/tools v0.9.1 // indirect
+	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
+	google.golang.org/protobuf v1.30.0 // indirect
+	gopkg.in/ini.v1 v1.67.0 // indirect
+	gopkg.in/nullbio/null.v6 v6.0.0-20161116030900-40264a2e6b79 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

internal/metrics/metrics.go

@@ -86,8 +86,13 @@ const (
 	errUnexpectedEOFStr  = "unexpected_eof_str"
 	errJSONSyntax        = "json_syntax"
 	errBlobTooBig        = "blob_too_big"
+	errInvalidPeerJSON   = "invalid_peer_json"
+	errInvalidPeerData   = "invalid_peer_data"
+	errRequestTooLarge   = "request_too_large"
 	errDeadlineExceeded  = "deadline_exceeded"
 	errHashMismatch      = "hash_mismatch"
+	errProtectedBlob     = "protected_blob"
+	errInvalidBlobHash   = "invalid_blob_hash"
 	errZeroByteBlob      = "zero_byte_blob"
 	errInvalidCharacter  = "invalid_character"
 	errBlobNotFound      = "blob_not_found"
@@ -296,10 +301,20 @@ func TrackError(direction string, e error) (shouldLog bool) { // shouldLog is a
 	} else if strings.Contains(err.Error(), "blob must be at most") {
 		//log.Warnln("blob must be at most X bytes is not the same as ErrBlobTooBig")
 		errType = errBlobTooBig
+	} else if strings.Contains(err.Error(), "invalid json request") {
+		errType = errInvalidPeerJSON
+	} else if strings.Contains(err.Error(), "Invalid data") {
+		errType = errInvalidPeerData
+	} else if strings.Contains(err.Error(), "request is too large") {
+		errType = errRequestTooLarge
+	} else if strings.Contains(err.Error(), "Invalid blob hash length") {
+		errType = errInvalidBlobHash
 	} else if strings.Contains(err.Error(), "hash of received blob data does not match hash from send request") {
 		errType = errHashMismatch
 	} else if strings.Contains(err.Error(), "blob not found") {
 		errType = errBlobNotFound
+	} else if strings.Contains(err.Error(), "requested blob is protected") {
+		errType = errProtectedBlob
 	} else if strings.Contains(err.Error(), "0-byte blob received") {
 		errType = errZeroByteBlob
 	} else if strings.Contains(err.Error(), "PROTOCOL_VIOLATION: tried to retire connection") {

lite_db/db.go

@@ -10,7 +10,7 @@ import (
 	"github.com/go-sql-driver/mysql"
 	_ "github.com/go-sql-driver/mysql" // blank import for db driver ensures its imported even if its not used
 	log "github.com/sirupsen/logrus"
-	"github.com/volatiletech/null"
+	"github.com/volatiletech/null/v8"
 )
 
 // SdBlob is a special blob that contains information on the rest of the blobs in the stream
@@ -111,7 +111,7 @@ func (s *SQL) HasBlob(hash string) (bool, error) {
 // HasBlobs checks if the database contains the set of blobs and returns a bool map.
 func (s *SQL) HasBlobs(hashes []string) (map[string]bool, error) {
 	exists, streamsNeedingTouch, err := s.hasBlobs(hashes)
-	s.touch(streamsNeedingTouch)
+	_ = s.touch(streamsNeedingTouch)
 	return exists, err
 }
 

meta/meta.go

@@ -6,9 +6,36 @@ import (
 	"time"
 )
 
-var Version = ""
-var Time = ""
-var BuildTime time.Time
+var (
+	name       = "prism-bin"
+	version    = "unknown"
+	commit     = "unknown"
+	commitLong = "unknown"
+	branch     = "unknown"
+	Time       = "unknown"
+	BuildTime  time.Time
+)
+
+// Name returns main application name
+func Name() string {
+	return name
+}
+
+// Version returns current application version
+func Version() string {
+	return version
+}
+
+// FullName returns current app version, commit and build time
+func FullName() string {
+	return fmt.Sprintf(
+		`Name: %v
+Version: %v
+branch: %v
+commit: %v
+commit long: %v
+build date: %v`, Name(), Version(), branch, commit, commitLong, BuildTime.String())
+}
 
 func init() {
 	if Time != "" {
@@ -20,11 +47,6 @@ func init() {
 }
 
 func VersionString() string {
-	version := Version
-	if version == "" {
-		version = "<unset>"
-	}
-
 	var buildTime string
 	if BuildTime.IsZero() {
 		buildTime = "<now>"

publish/publish.go

@@ -115,7 +115,7 @@ func Publish(client *lbrycrd.Client, path, name, address string, details Details
 	return signedTx, txid, nil
 }
 
-//TODO: lots of assumptions. hardcoded values need to be passed in or calculated
+// TODO: lots of assumptions. hardcoded values need to be passed in or calculated
 func baseTx(client *lbrycrd.Client, amount float64, changeAddress btcutil.Address) (*wire.MsgTx, error) {
 	txFee := 0.0002 // TODO: estimate this better?
 
@@ -222,8 +222,7 @@ func makeStream(path string) (stream.Stream, *pb.Stream, error) {
 	if err != nil {
 		return nil, nil, errors.Err(err)
 	}
-	defer file.Close()
-
+	defer func() { _ = file.Close() }()
 	enc := stream.NewEncoder(file)
 
 	s, err := enc.Stream()

readme.md

@@ -13,7 +13,7 @@ This code includes a Go implementations of the LBRY peer protocol, reflector pro
 
 - Install mysql 8 (5.7 might work too)
 - add a reflector user and database with password `reflector` with localhost access only
-- Create the tables as described [here](https://github.com/lbryio/reflector.go/blob/ittt/db/db.go#L735) (the link might not update as the code does so just look for the schema in that file)
+- Create the tables as described [here](https://github.com/lbryio/reflector.go/blob/master/db/db.go#L735) (the link might not update as the code does so just look for the schema in that file)
 
 #### We do not support running reflector.go as a blob receiver, however if you want to run it as a private blobcache you may compile it yourself and run it as following:
 ```bash
@@ -96,7 +96,7 @@ Flags:
 ```
 ## Running from Source
 
-This project requires [Go v1.16](https://golang.org/doc/install).
+This project requires [Go v1.20](https://golang.org/doc/install).
 
 On Ubuntu you can install it with `sudo snap install go --classic`
 
@@ -104,7 +104,7 @@ On Ubuntu you can install it with `sudo snap install go --classic`
 git clone git@github.com:lbryio/reflector.go.git
 cd reflector.go
 make
-./bin/prism-bin
+./dist/linux_amd64/prism-bin
 ```
 
 ## Contributing
@@ -118,7 +118,7 @@ This project is MIT licensed.
 ## Security
 
 We take security seriously. Please contact security@lbry.com regarding any security issues.
-Our PGP key is [here](https://keybase.io/lbry/key.asc) if you need it.
+Our PGP key is [here](https://lbry.com/faq/pgp-key) if you need it.
 
 ## Contact
 The primary contact for this project is [@Nikooo777](https://github.com/Nikooo777) (niko-at-lbry.com)

reflector/protected_content.go

@@ -0,0 +1,81 @@
+package reflector
+
+import (
+	"encoding/json"
+	"net/http"
+	"time"
+
+	"github.com/bluele/gcache"
+	"github.com/lbryio/lbry.go/v2/extras/errors"
+	"golang.org/x/sync/singleflight"
+)
+
+const protectedListURL = "https://api.odysee.com/file/list_protected"
+
+type ProtectedContent struct {
+	SDHash  string `json:"sd_hash"`
+	ClaimID string `json:"claim_id"`
+}
+
+var protectedCache = gcache.New(10).Expiration(2 * time.Minute).Build()
+
+func GetProtectedContent() (interface{}, error) {
+	cachedVal, err := protectedCache.Get("protected")
+	if err == nil && cachedVal != nil {
+		return cachedVal.(map[string]bool), nil
+	}
+
+	method := "GET"
+	var r struct {
+		Success bool               `json:"success"`
+		Error   string             `json:"error"`
+		Data    []ProtectedContent `json:"data"`
+	}
+
+	client := &http.Client{}
+	req, err := http.NewRequest(method, protectedListURL, nil)
+
+	if err != nil {
+		return nil, errors.Err(err)
+	}
+	res, err := client.Do(req)
+	if err != nil {
+		return nil, errors.Err(err)
+	}
+	defer func() { _ = res.Body.Close() }()
+	if res.StatusCode != http.StatusOK {
+		return nil, errors.Err("unexpected status code %d", res.StatusCode)
+	}
+	if err = json.NewDecoder(res.Body).Decode(&r); err != nil {
+		return nil, errors.Err(err)
+	}
+
+	if !r.Success {
+		return nil, errors.Prefix("file/list_protected API call", r.Error)
+	}
+
+	protectedMap := make(map[string]bool, len(r.Data))
+	for _, pc := range r.Data {
+		protectedMap[pc.SDHash] = true
+	}
+	err = protectedCache.Set("protected", protectedMap)
+	if err != nil {
+		return protectedMap, errors.Err(err)
+	}
+	return protectedMap, nil
+}
+
+var sf = singleflight.Group{}
+
+func IsProtected(sdHash string) bool {
+	val, err, _ := sf.Do("protected", GetProtectedContent)
+	if err != nil {
+		return false
+	}
+	cachedMap, ok := val.(map[string]bool)
+	if !ok {
+		return false
+	}
+
+	return cachedMap[sdHash]
+}

reflector/server.go

@@ -6,7 +6,6 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"io"
-	"io/ioutil"
 	"net"
 	"time"
 
@@ -62,7 +61,7 @@ func (s *Server) Shutdown() {
 	log.Println("reflector server stopped")
 }
 
-//Start starts the server to handle connections.
+// Start starts the server to handle connections.
 func (s *Server) Start(address string) error {
 	l, err := net.Listen(network, address)
 	if err != nil {
@@ -367,7 +366,7 @@ func (s *Server) read(conn net.Conn, v interface{}) error {
 	dec := json.NewDecoder(conn)
 	err = dec.Decode(v)
 	if err != nil {
-		data, _ := ioutil.ReadAll(dec.Buffered())
+		data, _ := io.ReadAll(dec.Buffered())
 		if len(data) > 0 {
 			return errors.Err("%s. Data: %s", err.Error(), hex.EncodeToString(data))
 		}

reflector/uploader.go

@@ -1,7 +1,6 @@
 package reflector
 
 import (
-	"io/ioutil"
 	"os"
 	"path"
 	"sync"
@@ -165,7 +164,7 @@ func (u *Uploader) uploadBlob(filepath string) (err error) {
 		}
 	}()
 
-	blob, err := ioutil.ReadFile(filepath)
+	blob, err := os.ReadFile(filepath)
 	if err != nil {
 		return errors.Err(err)
 	}

scripts/lint.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+
+err=0
+trap 'err=1' ERR
+# All the .go files, excluding auto generated folders
+GO_FILES=$(find . -iname '*.go' -type f)
+(
+	go install golang.org/x/tools/cmd/goimports@latest                   # Used in build script for generated files
+#	go install golang.org/x/lint/golint@latest                           # Linter
+	go install github.com/jgautheron/gocyclo@latest                      # Check against high complexity
+	go install github.com/mdempsky/unconvert@latest                      # Identifies unnecessary type conversions
+	go install github.com/kisielk/errcheck@latest                        # Checks for unhandled errors
+	go install github.com/opennota/check/cmd/varcheck@latest             # Checks for unused vars
+	go install github.com/opennota/check/cmd/structcheck@latest          # Checks for unused fields in structs
+)
+echo "Running varcheck..." && varcheck $(go list ./...)
+echo "Running structcheck..." && structcheck $(go list ./...)
+# go vet is the official Go static analyzer
+echo "Running go vet..." && go vet $(go list ./...)
+# checks for unhandled errors
+echo "Running errcheck..." && errcheck $(go list ./...)
+# check for unnecessary conversions - ignore autogen code
+echo "Running unconvert..." && unconvert -v $(go list ./...)
+echo "Running gocyclo..." && gocyclo -ignore "_test" -avg -over 28 $GO_FILES
+#echo "Running golint..." && golint -set_exit_status $(go list ./...)
+test $err = 0 # Return non-zero if any command failed

server/http/routes.go

@@ -6,12 +6,14 @@ import (
 	"time"
 
 	"github.com/lbryio/reflector.go/internal/metrics"
+	"github.com/lbryio/reflector.go/reflector"
 	"github.com/lbryio/reflector.go/shared"
 	"github.com/lbryio/reflector.go/store"
 
 	"github.com/lbryio/lbry.go/v2/extras/errors"
 
 	"github.com/gin-gonic/gin"
+	log "github.com/sirupsen/logrus"
 )
 
 func (s *Server) getBlob(c *gin.Context) {
@@ -22,9 +24,20 @@ func (s *Server) getBlob(c *gin.Context) {
 }
 
 func (s *Server) HandleGetBlob(c *gin.Context) {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Errorf("Recovered from panic: %v", r)
+		}
+	}()
 	start := time.Now()
 	hash := c.Query("hash")
+	edgeToken := c.Query("edge_token")
 
+	if reflector.IsProtected(hash) && edgeToken != s.edgeToken {
+		_ = c.Error(errors.Err("requested blob is protected"))
+		c.String(http.StatusForbidden, "requested blob is protected")
+		return
+	}
 	if s.missesCache.Has(hash) {
 		serialized, err := shared.NewBlobTrace(time.Since(start), "http").Serialize()
 		c.Header("Via", serialized)

server/http/server.go

@@ -21,15 +21,17 @@ type Server struct {
 	grp                *stop.Group
 	concurrentRequests int
 	missesCache        gcache.Cache
+	edgeToken          string
 }
 
 // NewServer returns an initialized Server pointer.
-func NewServer(store store.BlobStore, requestQueueSize int) *Server {
+func NewServer(store store.BlobStore, requestQueueSize int, edgeToken string) *Server {
 	return &Server{
 		store:              store,
 		grp:                stop.New(),
 		concurrentRequests: requestQueueSize,
 		missesCache:        gcache.New(2000).Expiration(5 * time.Minute).ARC().Build(),
+		edgeToken:          edgeToken,
 	}
 }
 
@@ -43,11 +45,12 @@ func (s *Server) Shutdown() {
 // Start starts the server listener to handle connections.
 func (s *Server) Start(address string) error {
 	gin.SetMode(gin.ReleaseMode)
-	router := gin.Default()
-	router.GET("/blob", s.getBlob)
-	router.HEAD("/blob", s.hasBlob)
+	router := gin.New()
+	router.Use(gin.Logger())
 	// Install nice.Recovery, passing the handler to call after recovery
 	router.Use(nice.Recovery(s.recoveryHandler))
+	router.GET("/blob", s.getBlob)
+	router.HEAD("/blob", s.hasBlob)
 	srv := &http.Server{
 		Addr:    address,
 		Handler: router,

server/http3/client.go

@@ -16,7 +16,7 @@ import (
 	"github.com/lbryio/lbry.go/v2/extras/errors"
 	"github.com/lbryio/lbry.go/v2/stream"
 
-	"github.com/lucas-clemente/quic-go/http3"
+	"github.com/quic-go/quic-go/http3"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -69,7 +69,7 @@ func (c *Client) HasBlob(hash string) (bool, error) {
 	if err != nil {
 		return false, errors.Err(err)
 	}
-	defer resp.Body.Close()
+	defer func() { _ = resp.Body.Close() }()
 	if resp.StatusCode == http.StatusOK {
 		return true, nil
 	}
@@ -86,7 +86,7 @@ func (c *Client) GetBlob(hash string) (stream.Blob, shared.BlobTrace, error) {
 	if err != nil {
 		return nil, shared.NewBlobTrace(time.Since(start), "http3"), errors.Err(err)
 	}
-	defer resp.Body.Close()
+	defer func() { _ = resp.Body.Close() }()
 
 	if resp.StatusCode == http.StatusNotFound {
 		fmt.Printf("%s blob not found %d\n", hash, resp.StatusCode)

server/http3/server.go

@@ -15,14 +15,15 @@ import (
 	"time"
 
 	"github.com/lbryio/reflector.go/internal/metrics"
+	"github.com/lbryio/reflector.go/reflector"
 	"github.com/lbryio/reflector.go/store"
 
 	"github.com/lbryio/lbry.go/v2/extras/errors"
 	"github.com/lbryio/lbry.go/v2/extras/stop"
 
 	"github.com/gorilla/mux"
-	"github.com/lucas-clemente/quic-go"
-	"github.com/lucas-clemente/quic-go/http3"
+	"github.com/quic-go/quic-go"
+	"github.com/quic-go/quic-go/http3"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -112,11 +113,9 @@ func (s *Server) Start(address string) error {
 		}
 	})
 	server := http3.Server{
-		Server: &http.Server{
-			Handler:   r,
-			Addr:      address,
-			TLSConfig: generateTLSConfig(),
-		},
+		Addr:       address,
+		Handler:    r,
+		TLSConfig:  generateTLSConfig(),
 		QuicConfig: quicConf,
 	}
 	go InitWorkers(s, s.concurrentRequests)
@@ -156,7 +155,7 @@ func generateTLSConfig() *tls.Config {
 
 func (s *Server) listenAndServe(server *http3.Server) {
 	err := server.ListenAndServe()
-	if err != nil && err.Error() != "server closed" {
+	if err != nil && err != quic.ErrServerClosed {
 		log.Errorln(errors.FullTrace(err))
 	}
 }
@@ -181,7 +180,10 @@ func (s *Server) HandleGetBlob(w http.ResponseWriter, r *http.Request) {
 			wantsTrace = false
 		}
 	}
-
+	if reflector.IsProtected(requestedBlob) {
+		http.Error(w, "requested blob is protected", http.StatusForbidden)
+		return
+	}
 	blob, trace, err := s.store.Get(requestedBlob)
 
 	if wantsTrace {

server/http3/store.go

@@ -14,8 +14,8 @@ import (
 	"github.com/lbryio/lbry.go/v2/extras/errors"
 	"github.com/lbryio/lbry.go/v2/stream"
 
-	"github.com/lucas-clemente/quic-go"
-	"github.com/lucas-clemente/quic-go/http3"
+	"github.com/quic-go/quic-go"
+	"github.com/quic-go/quic-go/http3"
 )
 
 // Store is a blob store that gets blobs from a peer.
@@ -74,7 +74,7 @@ func (p *Store) Has(hash string) (bool, error) {
 	if err != nil {
 		return false, err
 	}
-	defer c.Close()
+	defer func() { _ = c.Close() }()
 	return c.HasBlob(hash)
 }
 
@@ -93,7 +93,7 @@ func (p *Store) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 	if err != nil {
 		return nil, shared.NewBlobTrace(time.Since(start), p.Name()), err
 	}
-	defer c.Close()
+	defer func() { _ = c.Close() }()
 	return c.GetBlob(hash)
 }
 

server/peer/server.go

@@ -32,17 +32,15 @@ const (
 // Server is an instance of a peer server that houses the listener and store.
 type Server struct {
 	store  store.BlobStore
-	disk   *store.DiskStore
 	closed bool
 
 	grp *stop.Group
 }
 
 // NewServer returns an initialized Server pointer.
-func NewServer(blobstore store.BlobStore) *Server {
+func NewServer(store store.BlobStore) *Server {
 	return &Server{
-		store: blobstore,
-		disk:  store.NewDiskStore("/external/blobs", 2),
+		store: store,
 		grp:   stop.New(),
 	}
 }
@@ -133,7 +131,7 @@ func (s *Server) handleConnection(conn net.Conn) {
 			log.Error(errors.FullTrace(err))
 		}
 
-		response, blobHash, err := s.handleCompositeRequest(request)
+		response, err = s.handleCompositeRequest(request)
 		if err != nil {
 			log.Error(errors.FullTrace(err))
 			return
@@ -145,7 +143,6 @@ func (s *Server) handleConnection(conn net.Conn) {
 		}
 
 		n, err := conn.Write(response)
-		s.disk.Sendfile(blobHash, conn)
 		if err != nil {
 			if !strings.Contains(err.Error(), "connection reset by peer") { // means the other side closed the connection using TCP reset
 				s.logError(err)
@@ -224,70 +221,78 @@ func (s *Server) handleAvailabilityRequest(data []byte) ([]byte, error) {
 //	return append(response, blob...), nil
 //}
 
-func (s *Server) handleCompositeRequest(data []byte) ([]byte, string, error) {
+func (s *Server) handleCompositeRequest(data []byte) ([]byte, error) {
 	var request compositeRequest
 	err := json.Unmarshal(data, &request)
 	if err != nil {
 		var je *json.SyntaxError
 		if ee.As(err, &je) {
-			return nil, "", errors.Err("invalid json at offset %d in data %s", je.Offset, hex.EncodeToString(data))
+			return nil, errors.Err("invalid json request: offset %d in data %s", je.Offset, hex.EncodeToString(data))
 		}
-		return nil, "", errors.Err(err)
+		return nil, errors.Err(err)
 	}
 
 	response := compositeResponse{
 		LbrycrdAddress: LbrycrdAddress,
+		AvailableBlobs: []string{},
 	}
 
 	if len(request.RequestedBlobs) > 0 {
-		var availableBlobs []string
 		for _, blobHash := range request.RequestedBlobs {
+			if reflector.IsProtected(blobHash) {
+				return nil, errors.Err("requested blob is protected")
+			}
 			exists, err := s.store.Has(blobHash)
 			if err != nil {
-				return nil, "", err
+				return nil, err
 			}
 			if exists {
-				availableBlobs = append(availableBlobs, blobHash)
+				response.AvailableBlobs = append(response.AvailableBlobs, blobHash)
 			}
 		}
-		response.AvailableBlobs = availableBlobs
 	}
 
-	response.BlobDataPaymentRate = paymentRateAccepted
-	if request.BlobDataPaymentRate < 0 {
-		response.BlobDataPaymentRate = paymentRateTooLow
+	if request.BlobDataPaymentRate != nil {
+		response.BlobDataPaymentRate = paymentRateAccepted
+		if *request.BlobDataPaymentRate < 0 {
+			response.BlobDataPaymentRate = paymentRateTooLow
+		}
 	}
 
 	var blob []byte
+	var trace shared.BlobTrace
 	if request.RequestedBlob != "" {
 		if len(request.RequestedBlob) != stream.BlobHashHexLength {
-			return nil, "", errors.Err("Invalid blob hash length")
+			return nil, errors.Err("Invalid blob hash length")
 		}
 
 		log.Debugln("Sending blob " + request.RequestedBlob[:8])
 
-		size, err := s.disk.Size(request.RequestedBlob)
-		if err != nil {
-			response.IncomingBlob = incomingBlob{
+		blob, trace, err = s.store.Get(request.RequestedBlob)
+		log.Debug(trace.String())
+		if errors.Is(err, store.ErrBlobNotFound) {
+			response.IncomingBlob = &incomingBlob{
 				Error: err.Error(),
 			}
+		} else if err != nil {
+			return nil, err
 		} else {
-			response.IncomingBlob = incomingBlob{
+			response.IncomingBlob = &incomingBlob{
 				BlobHash: request.RequestedBlob,
-				Length:   int(size),
+				Length:   len(blob),
 			}
+			metrics.MtrOutBytesTcp.Add(float64(len(blob)))
+			metrics.BlobDownloadCount.Inc()
+			metrics.PeerDownloadCount.Inc()
 		}
-		metrics.MtrOutBytesTcp.Add(float64(len(blob)))
-		metrics.BlobDownloadCount.Inc()
-		metrics.PeerDownloadCount.Inc()
 	}
 
 	respData, err := json.Marshal(response)
 	if err != nil {
-		return nil, "", err
+		return nil, err
 	}
 
-	return respData, request.RequestedBlob, nil
+	return append(respData, blob...), nil
 }
 
 func (s *Server) logError(e error) {
@@ -301,7 +306,15 @@ func (s *Server) logError(e error) {
 }
 
 func readNextMessage(buf *bufio.Reader) ([]byte, error) {
-	msg := make([]byte, 0)
+	first_byte, err := buf.ReadByte()
+	if err != nil {
+		return nil, err
+	}
+	if first_byte != '{' {
+		// every request starts with '{'. Checking here disconnects earlier, so we don't wait until timeout
+		return nil, errInvalidData
+	}
+	msg := []byte("{")
 	eof := false
 
 	for {
@@ -322,6 +335,8 @@ func readNextMessage(buf *bufio.Reader) ([]byte, error) {
 
 			if len(msg) > maxRequestSize {
 				return msg, errRequestTooLarge
+			} else if len(msg) > 0 && msg[0] != '{' {
+				return msg, errInvalidData
 			}
 
 			// yes, this is how the peer protocol knows when the request finishes
@@ -356,6 +371,7 @@ const (
 )
 
 var errRequestTooLarge = errors.Base("request is too large")
+var errInvalidData = errors.Base("Invalid data")
 
 type availabilityRequest struct {
 	LbrycrdAddress bool     `json:"lbrycrd_address"`
@@ -392,13 +408,13 @@ type blobResponse struct {
 type compositeRequest struct {
 	LbrycrdAddress      bool     `json:"lbrycrd_address"`
 	RequestedBlobs      []string `json:"requested_blobs"`
-	BlobDataPaymentRate float64  `json:"blob_data_payment_rate"`
+	BlobDataPaymentRate *float64 `json:"blob_data_payment_rate"`
 	RequestedBlob       string   `json:"requested_blob"`
 }
 
 type compositeResponse struct {
-	LbrycrdAddress      string       `json:"lbrycrd_address,omitempty"`
-	AvailableBlobs      []string     `json:"available_blobs,omitempty"`
-	BlobDataPaymentRate string       `json:"blob_data_payment_rate,omitempty"`
-	IncomingBlob        incomingBlob `json:"incoming_blob,omitempty"`
+	LbrycrdAddress      string        `json:"lbrycrd_address,omitempty"`
+	AvailableBlobs      []string      `json:"available_blobs"`
+	BlobDataPaymentRate string        `json:"blob_data_payment_rate,omitempty"`
+	IncomingBlob        *incomingBlob `json:"incoming_blob,omitempty"`
 }

server/peer/server_test.go

@@ -2,7 +2,10 @@ package peer
 
 import (
 	"bytes"
+	"io"
+	"net"
 	"testing"
+	"time"
 
 	"github.com/lbryio/reflector.go/store"
 )
@@ -75,3 +78,62 @@ func TestAvailabilityRequest_WithBlobs(t *testing.T) {
 		}
 	}
 }
+
+func TestRequestFromConnection(t *testing.T) {
+	s := getServer(t, true)
+	err := s.Start("127.0.0.1:50505")
+	defer s.Shutdown()
+	if err != nil {
+		t.Error("error starting server", err)
+	}
+
+	for _, p := range availabilityRequests {
+		conn, err := net.Dial("tcp", "127.0.0.1:50505")
+		if err != nil {
+			t.Error("error opening connection", err)
+		}
+		defer func() { _ = conn.Close() }()
+
+		response := make([]byte, 8192)
+		_, err = conn.Write(p.request)
+		if err != nil {
+			t.Error("error writing", err)
+		}
+		_, err = conn.Read(response)
+		if err != nil {
+			t.Error("error reading", err)
+		}
+		if !bytes.Equal(response[:len(p.response)], p.response) {
+			t.Errorf("Response did not match expected response.\nExpected: %s\nGot: %s", string(p.response), string(response))
+		}
+	}
+}
+
+func TestInvalidData(t *testing.T) {
+	s := getServer(t, true)
+	err := s.Start("127.0.0.1:50503")
+	defer s.Shutdown()
+	if err != nil {
+		t.Error("error starting server", err)
+	}
+	conn, err := net.Dial("tcp", "127.0.0.1:50503")
+	if err != nil {
+		t.Error("error opening connection", err)
+	}
+	defer func() { _ = conn.Close() }()
+
+	response := make([]byte, 8192)
+	_, err = conn.Write([]byte("hello dear server, I would like blobs. Please"))
+	if err != nil {
+		t.Error("error writing", err)
+	}
+	err = conn.SetReadDeadline(time.Now().Add(5 * time.Second))
+	if err != nil {
+		t.Error("error setting read deadline", err)
+	}
+	_, err = conn.Read(response)
+	if err != io.EOF {
+		t.Error("error reading", err)
+	}
+	println(response)
+}

server/peer/store.go

@@ -42,7 +42,7 @@ func (p *Store) Has(hash string) (bool, error) {
 	if err != nil {
 		return false, err
 	}
-	defer c.Close()
+	defer func() { _ = c.Close() }()
 	return c.HasBlob(hash)
 }
 
@@ -53,7 +53,7 @@ func (p *Store) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 	if err != nil {
 		return nil, shared.NewBlobTrace(time.Since(start), p.Name()), err
 	}
-	defer c.Close()
+	defer func() { _ = c.Close() }()
 	blob, trace, err := c.GetBlob(hash)
 	if err != nil && strings.Contains(err.Error(), "blob not found") {
 		return nil, trace, store.ErrBlobNotFound

store/atime_linux.go

@@ -1,3 +1,4 @@
+//go:build linux
 // +build linux
 
 package store
@@ -9,7 +10,7 @@ import (
 )
 
 func timespecToTime(ts syscall.Timespec) time.Time {
-	return time.Unix(int64(ts.Sec), int64(ts.Nsec))
+	return time.Unix(ts.Sec, ts.Nsec)
 }
 
 func atime(fi os.FileInfo) time.Time {

store/cloudfront_ro.go

@@ -2,7 +2,6 @@ package store
 
 import (
 	"io"
-	"io/ioutil"
 	"net/http"
 	"time"
 
@@ -37,8 +36,7 @@ func (c *CloudFrontROStore) Has(hash string) (bool, error) {
 	if err != nil {
 		return false, err
 	}
-	defer body.Close()
-
+	defer func() { _ = body.Close() }()
 	switch status {
 	case http.StatusNotFound, http.StatusForbidden:
 		return false, nil
@@ -61,12 +59,12 @@ func (c *CloudFrontROStore) Get(hash string) (stream.Blob, shared.BlobTrace, err
 	if err != nil {
 		return nil, shared.NewBlobTrace(time.Since(start), c.Name()), err
 	}
-	defer body.Close()
+	defer func() { _ = body.Close() }()
 	switch status {
 	case http.StatusNotFound, http.StatusForbidden:
 		return nil, shared.NewBlobTrace(time.Since(start), c.Name()), errors.Err(ErrBlobNotFound)
 	case http.StatusOK:
-		b, err := ioutil.ReadAll(body)
+		b, err := io.ReadAll(body)
 		if err != nil {
 			return nil, shared.NewBlobTrace(time.Since(start), c.Name()), errors.Err(err)
 		}
@@ -83,7 +81,7 @@ func (c *CloudFrontROStore) cfRequest(method, hash string) (int, io.ReadCloser,
 	if err != nil {
 		return 0, nil, errors.Err(err)
 	}
-	req.Header.Add("User-Agent", "reflector.go/"+meta.Version)
+	req.Header.Add("User-Agent", "reflector.go/"+meta.Version())
 
 	res, err := http.DefaultClient.Do(req)
 	if err != nil {

store/dbbacked.go

@@ -112,22 +112,22 @@ func (d *DBBackedStore) Block(hash string) error {
 		return err
 	}
 
-	has, err := d.db.HasBlob(hash, false)
-	if err != nil {
-		return err
-	}
-
-	if has {
-		err = d.blobs.Delete(hash)
-		if err != nil {
-			return err
-		}
-
-		err = d.db.Delete(hash)
-		if err != nil {
-			return err
-		}
-	}
+	//has, err := d.db.HasBlob(hash, false)
+	//if err != nil {
+	//	return err
+	//}
+	//
+	//if has {
+	//	err = d.blobs.Delete(hash)
+	//	if err != nil {
+	//		return err
+	//	}
+	//
+	//	err = d.db.Delete(hash)
+	//	if err != nil {
+	//		return err
+	//	}
+	//}
 
 	return d.markBlocked(hash)
 }

store/disk.go

@@ -1,13 +1,6 @@
 package store
 
 import (
-	"bytes"
-	"crypto/sha512"
-	"encoding/hex"
-	"fmt"
-	"io"
-	"io/ioutil"
-	"net"
 	"os"
 	"path"
 	"time"
@@ -17,10 +10,6 @@ import (
 
 	"github.com/lbryio/lbry.go/v2/extras/errors"
 	"github.com/lbryio/lbry.go/v2/stream"
-
-	"github.com/brk0v/directio"
-	log "github.com/sirupsen/logrus"
-	"go.uber.org/atomic"
 )
 
 // DiskStore stores blobs on a local disk
@@ -32,12 +21,8 @@ type DiskStore struct {
 
 	// true if initOnce ran, false otherwise
 	initialized bool
-
-	concurrentChecks atomic.Int32
 }
 
-const maxConcurrentChecks = 30
-
 // NewDiskStore returns an initialized file disk store pointer.
 func NewDiskStore(dir string, prefixLength int) *DiskStore {
 	return &DiskStore{
@@ -68,14 +53,6 @@ func (d *DiskStore) Has(hash string) (bool, error) {
 	return true, nil
 }
 
-func (d *DiskStore) Size(hash string) (int64, error) {
-	stat, err := os.Stat(d.path(hash))
-	if err != nil {
-		return 0, err
-	}
-	return stat.Size(), nil
-}
-
 // Get returns the blob or an error if the blob doesn't exist.
 func (d *DiskStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 	start := time.Now()
@@ -84,69 +61,16 @@ func (d *DiskStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 		return nil, shared.NewBlobTrace(time.Since(start), d.Name()), err
 	}
 
-	blob, err := ioutil.ReadFile(d.path(hash))
+	blob, err := os.ReadFile(d.path(hash))
 	if err != nil {
 		if os.IsNotExist(err) {
 			return nil, shared.NewBlobTrace(time.Since(start), d.Name()), errors.Err(ErrBlobNotFound)
 		}
 		return nil, shared.NewBlobTrace(time.Since(start), d.Name()), errors.Err(err)
 	}
-
-	// this is a rather poor yet effective way of throttling how many blobs can be checked concurrently
-	// poor because there is a possible race condition between the check and the actual +1
-	if d.concurrentChecks.Load() < maxConcurrentChecks {
-		d.concurrentChecks.Add(1)
-		defer d.concurrentChecks.Sub(1)
-		hashBytes := sha512.Sum384(blob)
-		readHash := hex.EncodeToString(hashBytes[:])
-		if hash != readHash {
-			message := fmt.Sprintf("[%s] found a broken blob while reading from disk. Actual hash: %s", hash, readHash)
-			log.Errorf("%s", message)
-			err := d.Delete(hash)
-			if err != nil {
-				return nil, shared.NewBlobTrace(time.Since(start), d.Name()), err
-			}
-			return nil, shared.NewBlobTrace(time.Since(start), d.Name()), errors.Err(message)
-		}
-	}
-
 	return blob, shared.NewBlobTrace(time.Since(start), d.Name()), nil
 }
 
-// Put stores the blob on disk
-func (d *DiskStore) Put(hash string, blob stream.Blob) error {
-	err := d.initOnce()
-	if err != nil {
-		return err
-	}
-
-	err = d.ensureDirExists(d.dir(hash))
-	if err != nil {
-		return err
-	}
-
-	// Open file with O_DIRECT
-	f, err := os.OpenFile(d.tmpPath(hash), openFileFlags, 0644)
-	if err != nil {
-		return errors.Err(err)
-	}
-	defer f.Close()
-
-	// Use directio writer
-	dio, err := directio.New(f)
-	if err != nil {
-		return errors.Err(err)
-	}
-	defer dio.Flush()
-	// Write the body to file
-	_, err = io.Copy(dio, bytes.NewReader(blob))
-	if err != nil {
-		return errors.Err(err)
-	}
-	err = os.Rename(d.tmpPath(hash), d.path(hash))
-	return errors.Err(err)
-}
-
 // PutSD stores the sd blob on the disk
 func (d *DiskStore) PutSD(hash string, blob stream.Blob) error {
 	return d.Put(hash, blob)
@@ -220,15 +144,3 @@ func (d *DiskStore) initOnce() error {
 // Shutdown shuts down the store gracefully
 func (d *DiskStore) Shutdown() {
 }
-
-func (d *DiskStore) Sendfile(hash string, conn net.Conn) error {
-	f, err := os.Open(d.path(hash))
-	if err != nil {
-		return err
-	}
-	_, err = io.Copy(conn, f)
-	if err != nil {
-		return err
-	}
-	return nil
-}

store/disk_test.go

@@ -1,7 +1,6 @@
 package store
 
 import (
-	"io/ioutil"
 	"os"
 	"path"
 	"path/filepath"
@@ -14,9 +13,9 @@ import (
 )
 
 func TestDiskStore_Get(t *testing.T) {
-	tmpDir, err := ioutil.TempDir("", "reflector_test_*")
+	tmpDir, err := os.MkdirTemp("", "reflector_test_*")
 	require.NoError(t, err)
-	defer os.RemoveAll(tmpDir)
+	defer func() { _ = os.RemoveAll(tmpDir) }()
 	d := NewDiskStore(tmpDir, 2)
 
 	hash := "f428b8265d65dad7f8ffa52922bba836404cbd62f3ecfe10adba6b444f8f658938e54f5981ac4de39644d5b93d89a94b"
@@ -25,7 +24,7 @@ func TestDiskStore_Get(t *testing.T) {
 	expectedPath := path.Join(tmpDir, hash[:2], hash)
 	err = os.MkdirAll(filepath.Dir(expectedPath), os.ModePerm)
 	require.NoError(t, err)
-	err = ioutil.WriteFile(expectedPath, data, os.ModePerm)
+	err = os.WriteFile(expectedPath, data, os.ModePerm)
 	require.NoError(t, err)
 
 	blob, _, err := d.Get(hash)
@@ -34,9 +33,9 @@ func TestDiskStore_Get(t *testing.T) {
 }
 
 func TestDiskStore_GetNonexistentBlob(t *testing.T) {
-	tmpDir, err := ioutil.TempDir("", "reflector_test_*")
+	tmpDir, err := os.MkdirTemp("", "reflector_test_*")
 	require.NoError(t, err)
-	defer os.RemoveAll(tmpDir)
+	defer func() { _ = os.RemoveAll(tmpDir) }()
 	d := NewDiskStore(tmpDir, 2)
 
 	blob, _, err := d.Get("nonexistent")

store/diskstore_put_darwin.go

@@ -0,0 +1,42 @@
+//go:build darwin
+// +build darwin
+
+package store
+
+import (
+	"bytes"
+	"io"
+	"os"
+
+	"github.com/lbryio/lbry.go/v2/extras/errors"
+	"github.com/lbryio/lbry.go/v2/stream"
+)
+
+var openFileFlags = os.O_WRONLY | os.O_CREATE
+
+// Put stores the blob on disk
+func (d *DiskStore) Put(hash string, blob stream.Blob) error {
+	err := d.initOnce()
+	if err != nil {
+		return err
+	}
+
+	err = d.ensureDirExists(d.dir(hash))
+	if err != nil {
+		return err
+	}
+
+	// Open file with O_DIRECT
+	f, err := os.OpenFile(d.tmpPath(hash), openFileFlags, 0644)
+	if err != nil {
+		return errors.Err(err)
+	}
+	defer f.Close()
+
+	_, err = io.Copy(f, bytes.NewReader(blob))
+	if err != nil {
+		return errors.Err(err)
+	}
+	err = os.Rename(d.tmpPath(hash), d.path(hash))
+	return errors.Err(err)
+}

store/diskstore_put_linux.go

@@ -0,0 +1,49 @@
+//go:build linux
+// +build linux
+
+package store
+
+import (
+	"bytes"
+	"io"
+	"os"
+	"syscall"
+
+	"github.com/lbryio/lbry.go/v2/extras/errors"
+	"github.com/lbryio/lbry.go/v2/stream"
+
+	"github.com/brk0v/directio"
+)
+
+var openFileFlags = os.O_WRONLY | os.O_CREATE | syscall.O_DIRECT
+
+// Put stores the blob on disk
+func (d *DiskStore) Put(hash string, blob stream.Blob) error {
+	err := d.initOnce()
+	if err != nil {
+		return err
+	}
+
+	err = d.ensureDirExists(d.dir(hash))
+	if err != nil {
+		return err
+	}
+
+	// Open file with O_DIRECT
+	f, err := os.OpenFile(d.tmpPath(hash), openFileFlags, 0644)
+	if err != nil {
+		return errors.Err(err)
+	}
+	defer func() { _ = f.Close() }()
+	dio, err := directio.New(f)
+	if err != nil {
+		return errors.Err(err)
+	}
+	defer func() { _ = dio.Flush() }()
+	_, err = io.Copy(dio, bytes.NewReader(blob))
+	if err != nil {
+		return errors.Err(err)
+	}
+	err = os.Rename(d.tmpPath(hash), d.path(hash))
+	return errors.Err(err)
+}

store/flags_darwin.go

@@ -1,9 +0,0 @@
-// +build darwin
-
-package store
-
-import (
-	"os"
-)
-
-var openFileFlags = os.O_WRONLY | os.O_CREATE

store/flags_linux.go

@@ -1,10 +0,0 @@
-// +build linux
-
-package store
-
-import (
-	"os"
-	"syscall"
-)
-
-var openFileFlags = os.O_WRONLY | os.O_CREATE | syscall.O_DIRECT

store/gcache.go

@@ -59,7 +59,7 @@ func NewGcacheStore(component string, store BlobStore, maxSize int, strategy Evi
 	}
 	go func() {
 		if lstr, ok := store.(lister); ok {
-			err := l.loadExisting(lstr, int(maxSize))
+			err := l.loadExisting(lstr, maxSize)
 			if err != nil {
 				panic(err) // TODO: what should happen here? panic? return nil? just keep going?
 			}
@@ -144,11 +144,15 @@ func (l *GcacheStore) loadExisting(store lister, maxItems int) error {
 	logrus.Infof("read %d files from underlying store", len(existing))
 
 	added := 0
-	for _, h := range existing {
+	for i, h := range existing {
 		_ = l.cache.Set(h, true)
 		added++
 		if maxItems > 0 && added >= maxItems { // underlying cache is bigger than the cache
-			break
+			err := l.Delete(h)
+			logrus.Infof("deleted overflowing blob: %s (%d/%d)", h, i, len(existing))
+			if err != nil {
+				logrus.Warnf("error while deleting a blob that's overflowing the cache: %s", err.Error())
+			}
 		}
 	}
 	return nil

store/gcache_test.go

@@ -2,7 +2,6 @@ package store
 
 import (
 	"fmt"
-	"io/ioutil"
 	"os"
 	"reflect"
 	"testing"
@@ -88,9 +87,9 @@ func TestGcacheStore_UnderlyingBlobMissing(t *testing.T) {
 }
 
 func TestGcacheStore_loadExisting(t *testing.T) {
-	tmpDir, err := ioutil.TempDir("", "reflector_test_*")
+	tmpDir, err := os.MkdirTemp("", "reflector_test_*")
 	require.NoError(t, err)
-	defer os.RemoveAll(tmpDir)
+	defer func() { _ = os.RemoveAll(tmpDir) }()
 	d := NewDiskStore(tmpDir, 2)
 
 	hash := "hash"

store/http.go

@@ -2,8 +2,8 @@ package store
 
 import (
 	"bytes"
+	"context"
 	"io"
-	"io/ioutil"
 	"net"
 	"net/http"
 	"sync"
@@ -20,12 +20,14 @@ import (
 type HttpStore struct {
 	upstream   string
 	httpClient *http.Client
+	edgeToken  string
 }
 
-func NewHttpStore(upstream string) *HttpStore {
+func NewHttpStore(upstream, edgeToken string) *HttpStore {
 	return &HttpStore{
 		upstream:   "http://" + upstream,
 		httpClient: getClient(),
+		edgeToken:  edgeToken,
 	}
 }
 
@@ -44,7 +46,7 @@ func (n *HttpStore) Has(hash string) (bool, error) {
 	if err != nil {
 		return false, errors.Err(err)
 	}
-	defer res.Body.Close()
+	defer func() { _ = res.Body.Close() }()
 	if res.StatusCode == http.StatusNotFound {
 		return false, nil
 	}
@@ -53,7 +55,7 @@ func (n *HttpStore) Has(hash string) (bool, error) {
 	}
 	var body []byte
 	if res.Body != nil {
-		body, _ = ioutil.ReadAll(res.Body)
+		body, _ = io.ReadAll(res.Body)
 	}
 	return false, errors.Err("upstream error. Status code: %d (%s)", res.StatusCode, string(body))
 }
@@ -61,6 +63,9 @@ func (n *HttpStore) Has(hash string) (bool, error) {
 func (n *HttpStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 	start := time.Now()
 	url := n.upstream + "/blob?hash=" + hash
+	if n.edgeToken != "" {
+		url += "&edge_token=" + n.edgeToken
+	}
 
 	req, err := http.NewRequest("GET", url, nil)
 	if err != nil {
@@ -71,7 +76,7 @@ func (n *HttpStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 	if err != nil {
 		return nil, shared.NewBlobTrace(time.Since(start), n.Name()), errors.Err(err)
 	}
-	defer res.Body.Close()
+	defer func() { _ = res.Body.Close() }()
 	tmp := getBuffer()
 	defer putBuffer(tmp)
 	serialized := res.Header.Get("Via")
@@ -100,7 +105,7 @@ func (n *HttpStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 	}
 	var body []byte
 	if res.Body != nil {
-		body, _ = ioutil.ReadAll(res.Body)
+		body, _ = io.ReadAll(res.Body)
 	}
 
 	return nil, trace.Stack(time.Since(start), n.Name()), errors.Err("upstream error. Status code: %d (%s)", res.StatusCode, string(body))
@@ -138,14 +143,19 @@ func putBuffer(buf *bytes.Buffer) {
 	buffers.Put(buf)
 }
 
+func dialContext(ctx context.Context, network, address string) (net.Conn, error) {
+	dialer := &net.Dialer{
+		Timeout:   30 * time.Second,
+		KeepAlive: 30 * time.Second,
+	}
+	return dialer.DialContext(ctx, network, address)
+}
+
 // getClient gets an http client that's customized to be more performant when dealing with blobs of 2MB in size (most of our blobs)
 func getClient() *http.Client {
 	// Customize the Transport to have larger connection pool
 	defaultTransport := &http.Transport{
-		DialContext: (&net.Dialer{
-			Timeout:   30 * time.Second,
-			KeepAlive: 30 * time.Second,
-		}).DialContext,
+		DialContext:           dialContext,
 		ForceAttemptHTTP2:     true,
 		MaxIdleConns:          100,
 		IdleConnTimeout:       90 * time.Second,

store/speedwalk/speedwalk.go

@@ -1,7 +1,8 @@
 package speedwalk
 
 import (
-	"io/ioutil"
+	"io/fs"
+	"os"
 	"path/filepath"
 	"runtime"
 	"sync"
@@ -17,7 +18,19 @@ import (
 // AllFiles recursively lists every file in every subdirectory of a given directory
 // If basename is true, return the basename of each file. Otherwise return the full path starting at startDir.
 func AllFiles(startDir string, basename bool) ([]string, error) {
-	items, err := ioutil.ReadDir(startDir)
+	entries, err := os.ReadDir(startDir)
+	if err != nil {
+		return nil, err
+	}
+	items := make([]fs.FileInfo, 0, len(entries))
+	for _, entry := range entries {
+		info, err := entry.Info()
+		if err != nil {
+			return nil, err
+		}
+		items = append(items, info)
+	}
+
 	if err != nil {
 		return nil, err
 	}