From 7935624549104ccad3578eae9401073c6c5ddd0f Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Fri, 10 Dec 2021 15:21:29 -0600
Subject: [PATCH 01/23] implement percentiles on timestamp/decimal, still needs
 tests

---
 executor.go            | 33 ++++++++++++++++++++++++++++++---
 field.go               | 34 +++++++++++++++++++---------------
 field_internal_test.go | 16 ++++++++--------
 3 files changed, 57 insertions(+), 26 deletions(-)

diff --git a/executor.go b/executor.go
index 226759b55..90db6c74a 100644
--- a/executor.go
+++ b/executor.go
@@ -537,6 +537,11 @@ func (e *executor) execute(ctx context.Context, qcx *Qcx, index string, q *pql.Q
 			return nil, err
 		}
 
+		if vc, ok := v.(ValCount); ok {
+			vc.cleanup()
+			v = vc
+		}
+
 		results = append(results, v)
 		// Some Calls can have significant data associated with them
 		// that gets generated during processing, such as Precomputed
@@ -547,6 +552,22 @@ func (e *executor) execute(ctx context.Context, qcx *Qcx, index string, q *pql.Q
 	return results, nil
 }
 
+// cleanup removes the integer value (Val) from the ValCount if one of
+// the other fields is in use.
+//
+// ValCounts are normally holding data which is stored as a BSI
+// (integer) under the hood. Sometimes it's convenient to be able to
+// compare the underlying integer values rather than their
+// interpretation as decimal, timestamp, etc, so the lower level
+// functions may return both integer and the interpreted value, but we
+// don't want to pass that all the way back to the client, so we
+// remove it here.
+func (vc *ValCount) cleanup() {
+	if vc.Val != 0 && (vc.FloatVal != 0 || !vc.TimestampVal.IsZero() || vc.DecimalVal != nil) {
+		vc.Val = 0
+	}
+}
+
 // preprocessQuery expands any calls that need preprocessing.
 func (e *executor) preprocessQuery(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (*pql.Call, error) {
 	switch c.Name {
@@ -1211,6 +1232,10 @@ func (e *executor) executePercentile(ctx context.Context, qcx *Qcx, index string
 	if err != nil {
 		return ValCount{}, errors.New("Percentile(): field required")
 	}
+	field := e.Holder.Field(index, fieldName)
+	if field == nil {
+		return ValCount{}, ErrFieldNotFound
+	}
 
 	// filter call for min & max
 	var filterCall *pql.Call
@@ -1231,7 +1256,7 @@ func (e *executor) executePercentile(ctx context.Context, qcx *Qcx, index string
 		return ValCount{}, errors.Wrap(err, "executing Min call for Percentile")
 	}
 	if nthFloat == 0.0 {
-		return ValCount{Val: minVal.Val, Count: minVal.Count}, nil
+		return minVal, nil
 	}
 
 	// get max
@@ -1298,11 +1323,11 @@ func (e *executor) executePercentile(ctx context.Context, qcx *Qcx, index string
 		} else if leftCountWeighted < rightCount {
 			min = possibleNthVal + 1
 		} else {
-			return ValCount{Val: possibleNthVal, Count: 1}, nil
+			return field.valCountize(possibleNthVal, 1, nil)
 		}
 	}
 
-	return ValCount{Val: min, Count: 1}, nil
+	return field.valCountize(min, 1, nil)
 
 }
 
@@ -8057,6 +8082,8 @@ func getScaledInt(f *Field, v interface{}) (int64, error) {
 		switch tv := v.(type) {
 		case time.Time:
 			value = tv.UnixNano() / TimeUnitNanos(f.options.TimeUnit)
+		case int64:
+			value = tv
 		default:
 			return 0, errors.Errorf("unexpected timestamp value type %T, val %v", tv, tv)
 		}
diff --git a/field.go b/field.go
index 2aa865a1f..bcbc79c6e 100644
--- a/field.go
+++ b/field.go
@@ -1385,18 +1385,7 @@ func (f *Field) MaxForShard(tx Tx, shard uint64, filter *Row) (ValCount, error)
 		return ValCount{}, errors.Wrap(err, "calling fragment.max")
 	}
 
-	valCount := ValCount{Count: int64(cnt)}
-
-	if f.Options().Type == FieldTypeDecimal {
-		dec := pql.NewDecimal(max+bsig.Base, bsig.Scale)
-		valCount.DecimalVal = &dec
-	} else if f.Options().Type == FieldTypeTimestamp {
-		valCount.TimestampVal = time.Unix(0, (max+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC()
-	} else {
-		valCount.Val = max + bsig.Base
-	}
-
-	return valCount, nil
+	return f.valCountize(max, cnt, bsig)
 }
 
 // MinForShard returns the minimum value which appears in this shard
@@ -1431,6 +1420,23 @@ func (f *Field) MinForShard(tx Tx, shard uint64, filter *Row) (ValCount, error)
 		return ValCount{}, errors.Wrap(err, "calling fragment.min")
 	}
 
+	return f.valCountize(min, cnt, bsig)
+}
+
+// valCountize takes the "raw" min value and count we get from the
+// fragment and calculates the cooked values for this field
+// (timestamping, decimaling, or just adding in the base). It always
+// includes the int64 "Val\" value to make comparisons easier in the
+// executor (at time of writing, Percentile takes advantage of this,
+// but we might be able to simplify logic in other places as well).
+func (f *Field) valCountize(min int64, cnt uint64, bsig *bsiGroup) (ValCount, error) {
+	if bsig == nil {
+		bsig = f.bsiGroup(f.name)
+		if bsig == nil {
+			return ValCount{}, ErrBSIGroupNotFound
+		}
+
+	}
 	valCount := ValCount{Count: int64(cnt)}
 
 	if f.Options().Type == FieldTypeDecimal {
@@ -1438,10 +1444,8 @@ func (f *Field) MinForShard(tx Tx, shard uint64, filter *Row) (ValCount, error)
 		valCount.DecimalVal = &dec
 	} else if f.Options().Type == FieldTypeTimestamp {
 		valCount.TimestampVal = time.Unix(0, (min+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC()
-	} else {
-		valCount.Val = min + bsig.Base
 	}
-
+	valCount.Val = min + bsig.Base
 	return valCount, nil
 }
 
diff --git a/field_internal_test.go b/field_internal_test.go
index f1219f7e2..f7e0fadfd 100644
--- a/field_internal_test.go
+++ b/field_internal_test.go
@@ -748,29 +748,29 @@ func TestDecimalField_MinMaxForShard(t *testing.T) {
 			name:      "single",
 			columnIDs: []uint64{1},
 			values:    []float64{10.1},
-			expMax:    ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
-			expMin:    ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
+			expMax:    ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
+			expMin:    ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
 		},
 		{
 			name:      "twovals",
 			columnIDs: []uint64{1, 2},
 			values:    []float64{10.1, 20.2},
-			expMax:    ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 1},
-			expMin:    ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
+			expMax:    ValCount{Val: 20200, DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 1},
+			expMin:    ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
 		},
 		{
 			name:      "multiplecounts",
 			columnIDs: []uint64{1, 2, 3, 4, 5},
 			values:    []float64{10.1, 20.2, 10.1, 10.1, 20.2},
-			expMax:    ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2},
-			expMin:    ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3},
+			expMax:    ValCount{Val: 20200, DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2},
+			expMin:    ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3},
 		},
 		{
 			name:      "middlevals",
 			columnIDs: []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
 			values:    []float64{10.1, 20.2, 10.1, 10.1, 20.2, 11, 12, 11, 13, 11},
-			expMax:    ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2},
-			expMin:    ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3},
+			expMax:    ValCount{Val: 20200, DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2},
+			expMin:    ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3},
 		},
 	} {
 		t.Run(test.name+strconv.Itoa(i), func(t *testing.T) {

From 8486efaa79034398d8a27b38ed568d1145bae23f Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Mon, 13 Dec 2021 12:27:44 -0600
Subject: [PATCH 02/23] add exponential retry logic to internal http client,
 use in backup

---
 cmd/backup.go  | 11 ++++----
 ctl/backup.go  |  9 +++++--
 ctl/common.go  |  4 +--
 http/client.go | 68 ++++++++++++++++++++++++++++++++++++++++++++++----
 4 files changed, 78 insertions(+), 14 deletions(-)

diff --git a/cmd/backup.go b/cmd/backup.go
index 5761b6d57..133f0dbce 100644
--- a/cmd/backup.go
+++ b/cmd/backup.go
@@ -23,11 +23,12 @@ Backs up a FeatureBase server to a local, tar-formatted snapshot file.
 	}
 
 	flags := ccmd.Flags()
-	flags.StringVarP(&cmd.OutputDir, "output", "o", "", "output dir to write to")
-	flags.BoolVar(&cmd.NoSync, "no-sync", false, "disable file sync")
-	flags.IntVar(&cmd.Concurrency, "concurrency", cmd.Concurrency, "number of concurrent backup goroutines")
-	flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of FeatureBase.")
-	flags.StringVar(&cmd.Index, "index", "", "index to backup, default backs up all indexes. ")
+	flags.StringVarP(&cmd.OutputDir, "output", "o", "", "Output directory to write to.")
+	flags.BoolVar(&cmd.NoSync, "no-sync", false, "Disable file sync")
+	flags.IntVar(&cmd.Concurrency, "concurrency", cmd.Concurrency, "Number of concurrent backup goroutines.")
+	flags.StringVar(&cmd.Host, "host", "localhost:10101", "The address (host:port) of FeatureBase (HTTP).")
+	flags.StringVar(&cmd.Index, "index", "", "Index to backup, default backs up all indexes. ")
+	flags.DurationVar(&cmd.RetryPeriod, "retry-period", cmd.RetryPeriod, "Length of time after HTTP request failure to continue retrying request.")
 	ctl.SetTLSConfig(flags, "", &cmd.TLS.CertificatePath, &cmd.TLS.CertificateKeyPath, &cmd.TLS.CACertPath, &cmd.TLS.SkipVerify, &cmd.TLS.EnableClientVerification)
 	return ccmd
 }
diff --git a/ctl/backup.go b/ctl/backup.go
index 0e8a257f1..8ba57c278 100644
--- a/ctl/backup.go
+++ b/ctl/backup.go
@@ -10,6 +10,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"time"
 
 	pilosa "github.com/molecula/featurebase/v2"
 	"github.com/molecula/featurebase/v2/http"
@@ -37,6 +38,9 @@ type BackupCommand struct { // nolint: maligned
 	// Number of concurrent backup goroutines running at a time.
 	Concurrency int
 
+	// Amount of time after first failed request to continue retrying.
+	RetryPeriod time.Duration `json:"retry-period"`
+
 	// Reusable client.
 	client pilosa.InternalClient
 
@@ -51,6 +55,7 @@ func NewBackupCommand(stdin io.Reader, stdout, stderr io.Writer) *BackupCommand
 	return &BackupCommand{
 		CmdIO:       pilosa.NewCmdIO(stdin, stdout, stderr),
 		Concurrency: 1,
+		RetryPeriod: time.Minute,
 	}
 }
 
@@ -70,7 +75,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
 	}
 
 	// Create a client to the server.
-	client, err := commandClient(cmd)
+	client, err := commandClient(cmd, http.WithClientRetryPeriod(cmd.RetryPeriod))
 	if err != nil {
 		return fmt.Errorf("creating client: %w", err)
 	}
@@ -262,7 +267,7 @@ func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string,
 	logger := cmd.Logger()
 	logger.Printf("backing up shard: index=%q id=%d", indexName, shard)
 
-	client := http.NewInternalClientFromURI(&node.URI, http.GetHTTPClient(cmd.tlsConfig))
+	client := http.NewInternalClientFromURI(&node.URI, http.GetHTTPClient(cmd.tlsConfig), http.WithClientRetryPeriod(cmd.RetryPeriod))
 	rc, err := client.ShardReader(ctx, indexName, shard)
 	if err != nil {
 		return fmt.Errorf("fetching shard reader: %w", err)
diff --git a/ctl/common.go b/ctl/common.go
index c23b42f4c..7ba2df2f7 100644
--- a/ctl/common.go
+++ b/ctl/common.go
@@ -26,13 +26,13 @@ func SetTLSConfig(flags *pflag.FlagSet, prefix string, certificatePath *string,
 }
 
 // commandClient returns a pilosa.InternalHTTPClient for the command
-func commandClient(cmd CommandWithTLSSupport) (*http.InternalClient, error) {
+func commandClient(cmd CommandWithTLSSupport, opts ...http.InternalClientOption) (*http.InternalClient, error) {
 	tls := cmd.TLSConfiguration()
 	tlsConfig, err := server.GetTLSConfig(&tls, cmd.Logger())
 	if err != nil {
 		return nil, errors.Wrap(err, "getting tls config")
 	}
-	client, err := http.NewInternalClient(cmd.TLSHost(), http.GetHTTPClient(tlsConfig))
+	client, err := http.NewInternalClient(cmd.TLSHost(), http.GetHTTPClient(tlsConfig), opts...)
 	if err != nil {
 		return nil, errors.Wrap(err, "getting internal client")
 	}
diff --git a/http/client.go b/http/client.go
index 70c0639ec..8590a97fa 100644
--- a/http/client.go
+++ b/http/client.go
@@ -11,6 +11,7 @@ import (
 	"math/rand"
 	"net/http"
 	"net/url"
+	"os"
 	"path"
 	"sort"
 	"strconv"
@@ -20,6 +21,7 @@ import (
 	pilosa "github.com/molecula/featurebase/v2"
 	"github.com/molecula/featurebase/v2/encoding/proto"
 	"github.com/molecula/featurebase/v2/ingest"
+	"github.com/molecula/featurebase/v2/logger"
 	pnet "github.com/molecula/featurebase/v2/net"
 	"github.com/molecula/featurebase/v2/topology"
 	"github.com/molecula/featurebase/v2/tracing"
@@ -31,6 +33,10 @@ type InternalClient struct {
 	defaultURI *pnet.URI
 	serializer pilosa.Serializer
 
+	log logger.Logger
+
+	retryPeriod time.Duration
+
 	// The client to use for HTTP communication.
 	httpClient *http.Client
 	// the local node's API, used for operations that we can short-circuit that way
@@ -40,7 +46,7 @@ type InternalClient struct {
 // NewInternalClient returns a new instance of InternalClient to connect to host.
 // If api is non-nil, the client uses it for some same-host operations instead
 // of going through http.
-func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient, error) {
+func NewInternalClient(host string, remoteClient *http.Client, opts ...InternalClientOption) (*InternalClient, error) {
 	if host == "" {
 		return nil, pilosa.ErrHostRequired
 	}
@@ -50,16 +56,38 @@ func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient,
 		return nil, errors.Wrap(err, "getting URI")
 	}
 
-	client := NewInternalClientFromURI(uri, remoteClient)
+	client := NewInternalClientFromURI(uri, remoteClient, opts...)
 	return client, nil
 }
 
-func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client) *InternalClient {
-	return &InternalClient{
+type InternalClientOption func(c *InternalClient)
+
+// WithClientRetryPeriod is the max amount of total time the client will
+// retry failed requests using exponential backoff.
+func WithClientRetryPeriod(period time.Duration) InternalClientOption {
+	return func(c *InternalClient) {
+		c.retryPeriod = period
+	}
+}
+
+func WithClientLogger(log logger.Logger) InternalClientOption {
+	return func(c *InternalClient) {
+		c.log = log
+	}
+}
+
+func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, opts ...InternalClientOption) *InternalClient {
+	ic := &InternalClient{
 		defaultURI: defaultURI,
 		serializer: proto.Serializer{},
 		httpClient: remoteClient,
+		log:        logger.NewStandardLogger(os.Stderr),
 	}
+
+	for _, opt := range opts {
+		opt(ic)
+	}
+	return ic
 }
 
 // MaxShardByIndex returns the number of shards on a server by index.
@@ -1717,6 +1745,36 @@ func giveRawResponse(b bool) executeRequestOption {
 	}
 }
 
+func (c *InternalClient) doWithRetry(req *http.Request) (*http.Response, error) {
+	sleepDuration := time.Second
+	resp, err := c.httpClient.Do(req)
+	// start timer after first request, so if retryPeriod > 0 we
+	// pretty much always do at least one retry
+	start := time.Now()
+	for ; ; resp, err = c.httpClient.Do(req) {
+		if err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300 {
+			if time.Since(start) > c.retryPeriod {
+				break
+			}
+			if err != nil {
+				c.log.Printf("retrying request due to error: '%v'", err)
+			} else {
+				if bod, readErr := ioutil.ReadAll(resp.Body); readErr != nil {
+					c.log.Printf("retrying request due to status: %d, error reading body: '%v', body: '%s'", resp.StatusCode, readErr, bod)
+				} else {
+					c.log.Printf("retrying request due to status: %d, body: '%s'", resp.StatusCode, bod)
+				}
+			}
+
+			time.Sleep(sleepDuration)
+			sleepDuration *= 2
+		} else {
+			break
+		}
+	}
+	return resp, err
+}
+
 // executeRequest executes the given request and checks the Response. For
 // responses with non-2XX status, the body is read and closed, and an error is
 // returned. If the error is nil, the caller must ensure that the response body
@@ -1729,7 +1787,7 @@ func (c *InternalClient) executeRequest(req *http.Request, opts ...executeReques
 
 	tracing.GlobalTracer.InjectHTTPHeaders(req)
 	req.Close = false
-	resp, err := c.httpClient.Do(req)
+	resp, err := c.doWithRetry(req)
 	if err != nil {
 		if resp != nil {
 			resp.Body.Close()

From cdf4bc4c88272ca61dd9e51aee19e9298b5ab11d Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Mon, 13 Dec 2021 16:43:49 -0600
Subject: [PATCH 03/23] add clustertests testing backup's retry

---
 .circleci/config.yml                     |   2 +-
 Makefile                                 |   8 +-
 http/client.go                           |  32 +++---
 internal/clustertests/cluster_test.go    | 132 ++++++++++++++++-------
 internal/clustertests/pause_node_test.go |   7 +-
 5 files changed, 118 insertions(+), 63 deletions(-)

diff --git a/.circleci/config.yml b/.circleci/config.yml
index c3e1a43e4..f18532e49 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -151,7 +151,7 @@ jobs:
       - checkout-plus
       - skip-if-root-unchanged
       - setup_remote_docker
-      - run: make clustertests-build
+      - run: make clustertests
   release:
     executor:
       name: golang
diff --git a/Makefile b/Makefile
index a24d8c40a..e33c92b56 100644
--- a/Makefile
+++ b/Makefile
@@ -149,12 +149,10 @@ DOCKER_COMPOSE=internal/clustertests/docker-compose.yml
 clustertests: vendor
 	docker-compose -f $(DOCKER_COMPOSE) down
 	docker-compose -f $(DOCKER_COMPOSE) build
-	docker-compose -f $(DOCKER_COMPOSE) up --exit-code-from=client1
+	docker-compose -f $(DOCKER_COMPOSE) up -d pilosa1 pilosa2 pilosa3
+	docker-compose -f $(DOCKER_COMPOSE) run client1
+	docker-compose -f $(DOCKER_COMPOSE) down
 
-# Like clustertests, but rebuilds all images.
-clustertests-build: vendor
-	docker-compose -f $(DOCKER_COMPOSE) down -v
-	docker-compose -f $(DOCKER_COMPOSE) up --exit-code-from=client1 --build
 
 # Install Pilosa
 install:
diff --git a/http/client.go b/http/client.go
index 8590a97fa..bb08478d6 100644
--- a/http/client.go
+++ b/http/client.go
@@ -1751,26 +1751,22 @@ func (c *InternalClient) doWithRetry(req *http.Request) (*http.Response, error)
 	// start timer after first request, so if retryPeriod > 0 we
 	// pretty much always do at least one retry
 	start := time.Now()
-	for ; ; resp, err = c.httpClient.Do(req) {
-		if err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300 {
-			if time.Since(start) > c.retryPeriod {
-				break
-			}
-			if err != nil {
-				c.log.Printf("retrying request due to error: '%v'", err)
-			} else {
-				if bod, readErr := ioutil.ReadAll(resp.Body); readErr != nil {
-					c.log.Printf("retrying request due to status: %d, error reading body: '%v', body: '%s'", resp.StatusCode, readErr, bod)
-				} else {
-					c.log.Printf("retrying request due to status: %d, body: '%s'", resp.StatusCode, bod)
-				}
-			}
-
-			time.Sleep(sleepDuration)
-			sleepDuration *= 2
-		} else {
+	for ; err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300; resp, err = c.httpClient.Do(req) {
+		if time.Since(start) > c.retryPeriod {
 			break
 		}
+		if err != nil {
+			c.log.Printf("retrying request due to error: '%v'", err)
+		} else {
+			if bod, readErr := ioutil.ReadAll(resp.Body); readErr != nil {
+				c.log.Printf("retrying request due to status: %d, error reading body: '%v', body: '%s'", resp.StatusCode, readErr, bod)
+			} else {
+				c.log.Printf("retrying request due to status: %d, body: '%s'", resp.StatusCode, bod)
+			}
+		}
+
+		time.Sleep(sleepDuration)
+		sleepDuration *= 2
 	}
 	return resp, err
 }
diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go
index 2fca2f1ab..00967ac07 100644
--- a/internal/clustertests/cluster_test.go
+++ b/internal/clustertests/cluster_test.go
@@ -3,6 +3,8 @@ package clustertest
 
 import (
 	"context"
+	"fmt"
+	"io/ioutil"
 	"os"
 	"os/exec"
 	"testing"
@@ -30,56 +32,53 @@ func TestClusterStuff(t *testing.T) {
 		t.Fatalf("getting client: %v", err)
 	}
 
-	t.Run("long pause", func(t *testing.T) {
-		err := cli1.CreateIndex(context.Background(), "testidx", pilosa.IndexOptions{})
-		if err != nil {
-			t.Fatalf("creating index: %v", err)
-		}
-		err = cli1.CreateFieldWithOptions(context.Background(), "testidx", "testf", pilosa.FieldOptions{CacheType: pilosa.CacheTypeRanked, CacheSize: 100})
-		if err != nil {
-			t.Fatalf("creating field: %v", err)
-		}
+	if err := cli1.CreateIndex(context.Background(), "testidx", pilosa.IndexOptions{}); err != nil {
+		t.Fatalf("creating index: %v", err)
+	}
+	if err := cli1.CreateFieldWithOptions(context.Background(), "testidx", "testf", pilosa.FieldOptions{CacheType: pilosa.CacheTypeRanked, CacheSize: 100}); err != nil {
+		t.Fatalf("creating field: %v", err)
+	}
 
-		req := &pilosa.ImportRequest{
-			Index: "testidx",
-			Field: "testf",
-		}
-		req.ColumnIDs = make([]uint64, 10)
-		req.RowIDs = make([]uint64, 10)
+	req := &pilosa.ImportRequest{
+		Index: "testidx",
+		Field: "testf",
+	}
+	req.ColumnIDs = make([]uint64, 10)
+	req.RowIDs = make([]uint64, 10)
 
-		for i := 0; i < 1000; i++ {
-			req.RowIDs[i%10] = 0
-			req.ColumnIDs[i%10] = uint64((i/10)*pilosa.ShardWidth + i%10)
-			req.Shard = uint64(i / 10)
-			if i%10 == 9 {
-				err = cli1.Import(context.Background(), nil, req, &pilosa.ImportOptions{})
-				if err != nil {
-					t.Fatalf("importing: %v", err)
-				}
-			}
-		}
-
-		// Check query results from each node.
-		for i, cli := range []*picli.InternalClient{cli1, cli2, cli3} {
-			r, err := cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"})
+	for i := 0; i < 1000; i++ {
+		req.RowIDs[i%10] = 0
+		req.ColumnIDs[i%10] = uint64((i/10)*pilosa.ShardWidth + i%10)
+		req.Shard = uint64(i / 10)
+		if i%10 == 9 {
+			err = cli1.Import(context.Background(), nil, req, &pilosa.ImportOptions{})
 			if err != nil {
-				t.Fatalf("count querying pilosa%d: %v", i, err)
-			}
-			if r.Results[0].(uint64) != 1000 {
-				t.Fatalf("count on pilosa%d after import is %d", i, r.Results[0].(uint64))
+				t.Fatalf("importing: %v", err)
 			}
 		}
+	}
+
+	// Check query results from each node.
+	for i, cli := range []*picli.InternalClient{cli1, cli2, cli3} {
+		r, err := cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"})
+		if err != nil {
+			t.Fatalf("count querying pilosa%d: %v", i, err)
+		}
+		if r.Results[0].(uint64) != 1000 {
+			t.Fatalf("count on pilosa%d after import is %d", i, r.Results[0].(uint64))
+		}
+	}
+	t.Run("long pause", func(t *testing.T) {
 
 		pcmd := exec.Command("/pumba", "pause", "clustertests_pilosa3_1", "--duration", "10s")
 		pcmd.Stdout = os.Stdout
 		pcmd.Stderr = os.Stderr
 		t.Log("pausing pilosa3 for 10s")
-		err = pcmd.Start()
-		if err != nil {
+
+		if err := pcmd.Start(); err != nil {
 			t.Fatalf("starting pumba command: %v", err)
 		}
-		err = pcmd.Wait()
-		if err != nil {
+		if err := pcmd.Wait(); err != nil {
 			t.Fatalf("waiting on pumba pause cmd: %v", err)
 		}
 
@@ -98,6 +97,63 @@ func TestClusterStuff(t *testing.T) {
 			}
 		}
 	})
+
+	t.Run("backup", func(t *testing.T) {
+		// do backup with node 1 down, but restart it after a few seconds
+		if err := sendCmd("docker", "stop", "clustertests_pilosa1_1"); err != nil {
+			t.Fatalf("sending stop command: %v", err)
+		}
+		var backupCmd *exec.Cmd
+		tmpdir, err := ioutil.TempDir("", "")
+		if err != nil {
+			t.Fatalf("getting tmp dir: %v", err)
+		}
+		if backupCmd, err = startCmd(
+			"featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest")); err != nil {
+			t.Fatalf("sending backup command: %v", err)
+		}
+		time.Sleep(time.Second * 5)
+		if err = sendCmd("docker", "start", "clustertests_pilosa1_1"); err != nil {
+			t.Fatalf("sending start command: %v", err)
+		}
+
+		if err = backupCmd.Wait(); err != nil {
+			t.Fatalf("waiting on backup to finish: %v", err)
+		}
+
+		// now do backup with all nodes down and too short a timeout
+		// so it fails. Has be to be all 3 because the cluster has
+		// replicas=3 and the backup command will retry on replicas.
+		if err = sendCmd("docker", "stop", "clustertests_pilosa2_1"); err != nil {
+			t.Fatalf("sending stop command: %v", err)
+		}
+		if backupCmd, err = startCmd(
+			"featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest2"), "--retry-period=0.5s"); err != nil {
+			t.Fatalf("sending second backup command: %v", err)
+		}
+		time.Sleep(time.Millisecond * 5) // want the backup to get started, then fail
+		if err = sendCmd("docker", "stop", "clustertests_pilosa1_1"); err != nil {
+			t.Fatalf("sending stop command: %v", err)
+		}
+		if err = sendCmd("docker", "stop", "clustertests_pilosa3_1"); err != nil {
+			t.Fatalf("sending stop command: %v", err)
+		}
+
+		time.Sleep(time.Second * 5)
+		if err = sendCmd("docker", "start", "clustertests_pilosa1_1"); err != nil {
+			t.Fatalf("sending start command: %v", err)
+		}
+		if err = sendCmd("docker", "start", "clustertests_pilosa2_1"); err != nil {
+			t.Fatalf("sending start command: %v", err)
+		}
+		if err = sendCmd("docker", "start", "clustertests_pilosa3_1"); err != nil {
+			t.Fatalf("sending start command: %v", err)
+		}
+		if err = backupCmd.Wait(); err == nil {
+			t.Fatal("backup command should have errored but didn't")
+		}
+
+	})
 }
 
 func waitForStatus(t *testing.T, stator func(context.Context) (string, error), status string, n int, sleep time.Duration) {
diff --git a/internal/clustertests/pause_node_test.go b/internal/clustertests/pause_node_test.go
index 1cfa81417..256078a90 100644
--- a/internal/clustertests/pause_node_test.go
+++ b/internal/clustertests/pause_node_test.go
@@ -23,11 +23,16 @@ import (
 	"github.com/pkg/errors"
 )
 
-func sendCmd(cmd string, args ...string) error {
+func startCmd(cmd string, args ...string) (*exec.Cmd, error) {
 	pcmd := exec.Command(cmd, args...)
 	pcmd.Stdout = os.Stdout
 	pcmd.Stderr = os.Stderr
 	err := pcmd.Start()
+	return pcmd, err
+}
+
+func sendCmd(cmd string, args ...string) error {
+	pcmd, err := startCmd(cmd, args...)
 	if err != nil {
 		return errors.Wrap(err, "starting cmd")
 	}

From 3105a24542a6334608ee7bf82a59ecf66ab99a83 Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Mon, 20 Dec 2021 17:32:52 -0600
Subject: [PATCH 04/23] rewind Body on retry

this is really not ideal, and there are libraries for this kind of
thing, but I'd have to figure out how to make the libraries work with
everywhere we're already creating stdlib http clients.
---
 http/client.go | 20 ++++++++++++++++++++
 1 file changed, 20 insertions(+)

diff --git a/http/client.go b/http/client.go
index bb08478d6..0e2b1b428 100644
--- a/http/client.go
+++ b/http/client.go
@@ -1745,13 +1745,33 @@ func giveRawResponse(b bool) executeRequestOption {
 	}
 }
 
+type nopCloser struct {
+	*bytes.Reader
+}
+
+func (n nopCloser) Close() error {
+	return nil
+}
+
 func (c *InternalClient) doWithRetry(req *http.Request) (*http.Response, error) {
 	sleepDuration := time.Second
+	newBody := nopCloser{}
+	if req.Body != nil {
+		bod, err := ioutil.ReadAll(req.Body)
+		if err != nil {
+			return nil, errors.Wrap(err, "reading body")
+		}
+		newBody.Reader = bytes.NewReader(bod)
+		req.Body = newBody
+	}
 	resp, err := c.httpClient.Do(req)
 	// start timer after first request, so if retryPeriod > 0 we
 	// pretty much always do at least one retry
 	start := time.Now()
 	for ; err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300; resp, err = c.httpClient.Do(req) {
+		if newBody.Reader != nil {
+			newBody.Seek(0, io.SeekStart)
+		}
 		if time.Since(start) > c.retryPeriod {
 			break
 		}

From f676fbfc5151c68478c46c9f18da97ede2ac2967 Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Mon, 20 Dec 2021 17:42:22 -0600
Subject: [PATCH 05/23] add retryability to restore command

---
 cmd/restore.go |  1 +
 ctl/restore.go | 21 ++++++++++++++++-----
 go.mod         |  1 +
 go.sum         |  5 +++++
 4 files changed, 23 insertions(+), 5 deletions(-)

diff --git a/cmd/restore.go b/cmd/restore.go
index e9af62d24..71fb5ec7d 100644
--- a/cmd/restore.go
+++ b/cmd/restore.go
@@ -25,6 +25,7 @@ The Restore command will take a backup archive and restore it to a new, clean cl
 	flags.StringVarP(&cmd.Path, "source", "s", "", "backup file; specify '-' to restore from stdin tar stream")
 	flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of FeatureBase.")
 	flags.IntVar(&cmd.Concurrency, "concurrency", 1, "number of concurrent uploads")
+	flags.DurationVar(&cmd.RetryPeriod, "retry-period", cmd.RetryPeriod, "Length of time after HTTP request failure to continue retrying request.")
 	ctl.SetTLSConfig(
 		flags, "",
 		&cmd.TLS.CertificatePath,
diff --git a/ctl/restore.go b/ctl/restore.go
index 373581d5d..0804c3fa2 100644
--- a/ctl/restore.go
+++ b/ctl/restore.go
@@ -13,8 +13,12 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"time"
+
+	"github.com/hashicorp/go-retryablehttp"
 
 	pilosa "github.com/molecula/featurebase/v2"
+	fb_http "github.com/molecula/featurebase/v2/http"
 	"github.com/molecula/featurebase/v2/server"
 	"github.com/molecula/featurebase/v2/topology"
 	"golang.org/x/sync/errgroup"
@@ -29,6 +33,10 @@ type RestoreCommand struct {
 
 	// Filepath to the backup file.
 	Path string
+
+	// Amount of time after first failed request to continue retrying.
+	RetryPeriod time.Duration `json:"retry-period"`
+
 	// Reusable client.
 	client pilosa.InternalClient
 
@@ -62,7 +70,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
 		return fmt.Errorf("parsing tls config: %w", err)
 	}
 	// Create a client to the server.
-	client, err := commandClient(cmd)
+	client, err := commandClient(cmd, fb_http.WithClientRetryPeriod(cmd.RetryPeriod))
 	if err != nil {
 		return fmt.Errorf("creating client: %w", err)
 	}
@@ -119,7 +127,8 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.
 	if len(existingSchema) == 0 {
 		cmd.Logger().Printf("Load Schema")
 		url := primary.URI.Path("/schema")
-		var client http.Client
+		client := retryablehttp.NewClient()
+		client.RetryWaitMax = cmd.RetryPeriod
 		_, err = client.Post(url, "application/json", f)
 	} else {
 		schema := &pilosa.Schema{}
@@ -174,7 +183,8 @@ func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology
 	logger.Printf("Load idalloc")
 	url := primary.URI.Path("/internal/idalloc/restore")
 
-	var client http.Client
+	client := retryablehttp.NewClient()
+	client.RetryWaitMax = cmd.RetryPeriod
 	_, err = client.Post(url, "application/octet-stream", f)
 	return err
 }
@@ -244,14 +254,15 @@ func (cmd *RestoreCommand) restoreShard(ctx context.Context, filename string) er
 		defer f.Close()
 
 		url := node.URI.Path(fmt.Sprintf("/internal/restore/%v/%v", indexName, shard))
-		req, err := http.NewRequest("POST", url, f)
+		req, err := retryablehttp.NewRequest("POST", url, f)
 		if err != nil {
 			return err
 		}
 		req = req.WithContext(ctx)
 		req.Header.Set("Content-Type", "application/octet-stream")
 
-		var client http.Client
+		client := retryablehttp.NewClient()
+		client.RetryWaitMax = cmd.RetryPeriod
 		resp, err := client.Do(req)
 		if err != nil {
 			return err
diff --git a/go.mod b/go.mod
index b66980688..58b7e60c2 100644
--- a/go.mod
+++ b/go.mod
@@ -25,6 +25,7 @@ require (
 	github.com/gopherjs/gopherjs v0.0.0-20200217142428-fce0ec30dd00 // indirect
 	github.com/gorilla/handlers v1.3.0
 	github.com/gorilla/mux v1.7.0
+	github.com/hashicorp/go-retryablehttp v0.7.0
 	github.com/improbable-eng/grpc-web v0.13.0
 	github.com/lib/pq v1.8.0
 	github.com/molecula/apophenia v0.0.0-20190827192002-68b7a14a478b
diff --git a/go.sum b/go.sum
index 4224f965d..a60699bb5 100644
--- a/go.sum
+++ b/go.sum
@@ -186,10 +186,15 @@ github.com/grpc-ecosystem/grpc-gateway v1.9.5/go.mod h1:vNeuVxBJEsws4ogUvrchl83t
 github.com/hashicorp/consul/api v1.1.0/go.mod h1:VmuI/Lkw1nC05EYQWNKwWGbkg+FbDBtguAZLlVdkD9Q=
 github.com/hashicorp/consul/sdk v0.1.1/go.mod h1:VKf9jXwCTEY1QZP2MOLRhb5i/I/ssyNV1vwHyQBF0x8=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/go-cleanhttp v0.5.1 h1:dH3aiDG9Jvb5r5+bYHsikaOUIpcM0xvgMXVoDkXMzJM=
 github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
+github.com/hashicorp/go-hclog v0.9.2 h1:CG6TE5H9/JXsFWJCfoIVpKFIkFe6ysEuHirp4DxCsHI=
+github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
 github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
 github.com/hashicorp/go-msgpack v0.5.3/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM=
 github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
+github.com/hashicorp/go-retryablehttp v0.7.0 h1:eu1EI/mbirUgP5C8hVsTNaGZreBDlYiwC1FZWkvQPQ4=
+github.com/hashicorp/go-retryablehttp v0.7.0/go.mod h1:vAew36LZh98gCBJNLH42IQ1ER/9wtLZZ8meHqQvEYWY=
 github.com/hashicorp/go-rootcerts v1.0.0/go.mod h1:K6zTfqpRlCUIjkwsN4Z+hiSfzSTQa6eBIzfwKfwNnHU=
 github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU=
 github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=

From d3b9193c8d2cf29b135910b0990490d42c86b95f Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Mon, 20 Dec 2021 21:48:45 -0600
Subject: [PATCH 06/23] try to fix data race with http lib

WARNING: DATA RACE
Write at 0x00c008121e80 by goroutine 235:
  bytes.(*Reader).WriteTo()
      /usr/local/go/src/bytes/reader.go:139 +0x45
  github.com/molecula/featurebase/v2/http.nopCloser.WriteTo()
      <autogenerated>:1 +0x5d
  io.copyBuffer()
      /usr/local/go/src/io/io.go:391 +0x482
  io.Copy()
      /usr/local/go/src/io/io.go:368 +0x78
  net/http.(*transferWriter).doBodyCopy()
      /usr/local/go/src/net/http/transfer.go:400 +0x2f
  net/http.(*transferWriter).writeBody()
      /usr/local/go/src/net/http/transfer.go:364 +0xc9a
  net/http.(*Request).write()
      /usr/local/go/src/net/http/request.go:682 +0x887
  net/http.(*persistConn).writeLoop()
      /usr/local/go/src/net/http/transport.go:2343 +0x349

Previous write at 0x00c008121e80 by goroutine 192:
  bytes.(*Reader).Seek()
      /usr/local/go/src/bytes/reader.go:118 +0x824
  github.com/molecula/featurebase/v2/http.(*InternalClient).doWithRetry()
      /go/src/github.com/molecula/featurebase/http/client.go:1773 +0x86d
  github.com/molecula/featurebase/v2/http.(*InternalClient).executeRequest()
      /go/src/github.com/molecula/featurebase/http/client.go:1806 +0x15b
  github.com/molecula/featurebase/v2/http.(*InternalClient).CreateIndex()
      /go/src/github.com/molecula/featurebase/http/client.go:433 +0xbf8
  github.com/molecula/featurebase/v2/server_test.TestMain_Set_Quick.func1()
      /go/src/github.com/molecula/featurebase/server/server_test.go:64 +0x624
  testing.tRunner()
      /usr/local/go/src/testing/testing.go:1123 +0x202

Goroutine 235 (running) created at:
  net/http.(*Transport).dialConn()
      /usr/local/go/src/net/http/transport.go:1709 +0xc30
  net/http.(*Transport).dialConnFor()
      /usr/local/go/src/net/http/transport.go:1421 +0x151

Goroutine 192 (running) created at:
  testing.(*T).Run()
      /usr/local/go/src/testing/testing.go:1168 +0x5bb
  github.com/molecula/featurebase/v2/server_test.TestMain_Set_Quick()
      /go/src/github.com/molecula/featurebase/server/server_test.go:45 +0x116
  testing.tRunner()
      /usr/local/go/src/testing/testing.go:1123 +0x202
---
 http/client.go | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/http/client.go b/http/client.go
index 0e2b1b428..c32dd86ae 100644
--- a/http/client.go
+++ b/http/client.go
@@ -1755,22 +1755,22 @@ func (n nopCloser) Close() error {
 
 func (c *InternalClient) doWithRetry(req *http.Request) (*http.Response, error) {
 	sleepDuration := time.Second
-	newBody := nopCloser{}
+	var bod []byte
+	var err error
 	if req.Body != nil {
-		bod, err := ioutil.ReadAll(req.Body)
+		bod, err = ioutil.ReadAll(req.Body)
 		if err != nil {
 			return nil, errors.Wrap(err, "reading body")
 		}
-		newBody.Reader = bytes.NewReader(bod)
-		req.Body = newBody
+		req.Body = nopCloser{bytes.NewReader(bod)}
 	}
 	resp, err := c.httpClient.Do(req)
 	// start timer after first request, so if retryPeriod > 0 we
 	// pretty much always do at least one retry
 	start := time.Now()
 	for ; err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300; resp, err = c.httpClient.Do(req) {
-		if newBody.Reader != nil {
-			newBody.Seek(0, io.SeekStart)
+		if req.Body != nil {
+			req.Body = nopCloser{bytes.NewReader(bod)} // can't seek due to races with http lib internals
 		}
 		if time.Since(start) > c.retryPeriod {
 			break

From 2bce39644554b3a7eee8ec6af23511ebd55f64d2 Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Mon, 20 Dec 2021 22:16:06 -0600
Subject: [PATCH 07/23] add retry restore test and custom retry policy

---
 ctl/restore.go                        | 10 ++++++++
 internal/clustertests/cluster_test.go | 35 +++++++++++++++++++++++----
 2 files changed, 40 insertions(+), 5 deletions(-)

diff --git a/ctl/restore.go b/ctl/restore.go
index 0804c3fa2..cf99619cb 100644
--- a/ctl/restore.go
+++ b/ctl/restore.go
@@ -49,6 +49,7 @@ type RestoreCommand struct {
 func NewRestoreCommand(stdin io.Reader, stdout, stderr io.Writer) *RestoreCommand {
 	return &RestoreCommand{
 		CmdIO:       pilosa.NewCmdIO(stdin, stdout, stderr),
+		RetryPeriod: time.Second * 30,
 		Concurrency: 1,
 	}
 }
@@ -168,6 +169,13 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.
 	return err
 }
 
+func RetryWith400(ctx context.Context, resp *http.Response, err error) (bool, error) {
+	if resp != nil && resp.StatusCode > 400 { // we have some dumb status codes
+		return true, nil
+	}
+	return retryablehttp.DefaultRetryPolicy(ctx, resp, err)
+}
+
 func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology.Node) error {
 	logger := cmd.Logger()
 
@@ -185,6 +193,7 @@ func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology
 
 	client := retryablehttp.NewClient()
 	client.RetryWaitMax = cmd.RetryPeriod
+	client.CheckRetry = RetryWith400
 	_, err = client.Post(url, "application/octet-stream", f)
 	return err
 }
@@ -263,6 +272,7 @@ func (cmd *RestoreCommand) restoreShard(ctx context.Context, filename string) er
 
 		client := retryablehttp.NewClient()
 		client.RetryWaitMax = cmd.RetryPeriod
+		client.CheckRetry = RetryWith400
 		resp, err := client.Do(req)
 		if err != nil {
 			return err
diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go
index 00967ac07..af89abc2a 100644
--- a/internal/clustertests/cluster_test.go
+++ b/internal/clustertests/cluster_test.go
@@ -4,7 +4,7 @@ package clustertest
 import (
 	"context"
 	"fmt"
-	"io/ioutil"
+	"net/http"
 	"os"
 	"os/exec"
 	"testing"
@@ -104,10 +104,7 @@ func TestClusterStuff(t *testing.T) {
 			t.Fatalf("sending stop command: %v", err)
 		}
 		var backupCmd *exec.Cmd
-		tmpdir, err := ioutil.TempDir("", "")
-		if err != nil {
-			t.Fatalf("getting tmp dir: %v", err)
-		}
+		tmpdir := t.TempDir()
 		if backupCmd, err = startCmd(
 			"featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest")); err != nil {
 			t.Fatalf("sending backup command: %v", err)
@@ -121,6 +118,34 @@ func TestClusterStuff(t *testing.T) {
 			t.Fatalf("waiting on backup to finish: %v", err)
 		}
 
+		fmt.Println("STARTING RESTORE")
+
+		client := http.Client{}
+		if req, err := http.NewRequest(http.MethodDelete, "http://pilosa1:10101/index/testidx", nil); err != nil {
+			t.Fatalf("getting req: %v", err)
+		} else if resp, err := client.Do(req); err != nil {
+			t.Fatalf("doing request: %v", err)
+		} else if resp.StatusCode >= 400 {
+			t.Fatalf("bad response: %v", resp)
+		}
+
+		var restoreCmd *exec.Cmd
+		if restoreCmd, err = startCmd("featurebase", "restore", "-s", tmpdir+"/backuptest", "--host", "pilosa1:10101"); err != nil {
+			t.Fatalf("starting restore: %v", err)
+		}
+		time.Sleep(time.Millisecond * 50)
+		if err = sendCmd("docker", "stop", "clustertests_pilosa2_1"); err != nil {
+			t.Fatalf("sending stop command: %v", err)
+		}
+
+		time.Sleep(time.Second * 10)
+		if err = sendCmd("docker", "start", "clustertests_pilosa2_1"); err != nil {
+			t.Fatalf("sending stop command: %v", err)
+		}
+		if err := restoreCmd.Wait(); err != nil {
+			t.Fatalf("restore failed: %v", err)
+		}
+
 		// now do backup with all nodes down and too short a timeout
 		// so it fails. Has be to be all 3 because the cluster has
 		// replicas=3 and the backup command will retry on replicas.

From cde3f6b5ea5744135410f300e92acc895912f756 Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Tue, 21 Dec 2021 15:34:13 -0600
Subject: [PATCH 08/23] add profiling to backup/restore

---
 cmd/backup.go   |  1 +
 cmd/restore.go  |  1 +
 ctl/backup.go   | 18 +++++++++++++---
 ctl/restore.go  | 11 +++++++++-
 ctl/util.go     | 56 +++++++++++++++++++++++++++++++++++++++++++++++++
 http/handler.go |  2 +-
 6 files changed, 84 insertions(+), 5 deletions(-)
 create mode 100644 ctl/util.go

diff --git a/cmd/backup.go b/cmd/backup.go
index 133f0dbce..28712123d 100644
--- a/cmd/backup.go
+++ b/cmd/backup.go
@@ -29,6 +29,7 @@ Backs up a FeatureBase server to a local, tar-formatted snapshot file.
 	flags.StringVar(&cmd.Host, "host", "localhost:10101", "The address (host:port) of FeatureBase (HTTP).")
 	flags.StringVar(&cmd.Index, "index", "", "Index to backup, default backs up all indexes. ")
 	flags.DurationVar(&cmd.RetryPeriod, "retry-period", cmd.RetryPeriod, "Length of time after HTTP request failure to continue retrying request.")
+	flags.StringVar(&cmd.Pprof, "pprof", cmd.Pprof, "host:port to listen for profiling requests at /debug/pprof and /debug/fgprof.")
 	ctl.SetTLSConfig(flags, "", &cmd.TLS.CertificatePath, &cmd.TLS.CertificateKeyPath, &cmd.TLS.CACertPath, &cmd.TLS.SkipVerify, &cmd.TLS.EnableClientVerification)
 	return ccmd
 }
diff --git a/cmd/restore.go b/cmd/restore.go
index 71fb5ec7d..bc9271d0e 100644
--- a/cmd/restore.go
+++ b/cmd/restore.go
@@ -26,6 +26,7 @@ The Restore command will take a backup archive and restore it to a new, clean cl
 	flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of FeatureBase.")
 	flags.IntVar(&cmd.Concurrency, "concurrency", 1, "number of concurrent uploads")
 	flags.DurationVar(&cmd.RetryPeriod, "retry-period", cmd.RetryPeriod, "Length of time after HTTP request failure to continue retrying request.")
+	flags.StringVar(&cmd.Pprof, "pprof", cmd.Pprof, "host:port to listen for profiling requests at /debug/pprof and /debug/fgprof.")
 	ctl.SetTLSConfig(
 		flags, "",
 		&cmd.TLS.CertificatePath,
diff --git a/ctl/backup.go b/ctl/backup.go
index 8ba57c278..302041dfe 100644
--- a/ctl/backup.go
+++ b/ctl/backup.go
@@ -13,9 +13,10 @@ import (
 	"time"
 
 	pilosa "github.com/molecula/featurebase/v2"
-	"github.com/molecula/featurebase/v2/http"
+	fb_http "github.com/molecula/featurebase/v2/http"
 	"github.com/molecula/featurebase/v2/server"
 	"github.com/molecula/featurebase/v2/topology"
+	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -41,6 +42,9 @@ type BackupCommand struct { // nolint: maligned
 	// Amount of time after first failed request to continue retrying.
 	RetryPeriod time.Duration `json:"retry-period"`
 
+	// Host:port on which to listen for pprof.
+	Pprof string `json:"pprof"`
+
 	// Reusable client.
 	client pilosa.InternalClient
 
@@ -56,11 +60,19 @@ func NewBackupCommand(stdin io.Reader, stdout, stderr io.Writer) *BackupCommand
 		CmdIO:       pilosa.NewCmdIO(stdin, stdout, stderr),
 		Concurrency: 1,
 		RetryPeriod: time.Minute,
+		Pprof:       "localhost:43809",
 	}
 }
 
 // Run executes the main program execution.
 func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
+	logger := cmd.Logger()
+	close, err := startProfilingServer(cmd.Pprof, logger)
+	if err != nil {
+		return errors.Wrap(err, "starting profiling server")
+	}
+	defer close()
+
 	// Validate arguments.
 	if cmd.OutputDir == "" {
 		return fmt.Errorf("-o flag required")
@@ -75,7 +87,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
 	}
 
 	// Create a client to the server.
-	client, err := commandClient(cmd, http.WithClientRetryPeriod(cmd.RetryPeriod))
+	client, err := commandClient(cmd, fb_http.WithClientRetryPeriod(cmd.RetryPeriod))
 	if err != nil {
 		return fmt.Errorf("creating client: %w", err)
 	}
@@ -267,7 +279,7 @@ func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string,
 	logger := cmd.Logger()
 	logger.Printf("backing up shard: index=%q id=%d", indexName, shard)
 
-	client := http.NewInternalClientFromURI(&node.URI, http.GetHTTPClient(cmd.tlsConfig), http.WithClientRetryPeriod(cmd.RetryPeriod))
+	client := fb_http.NewInternalClientFromURI(&node.URI, fb_http.GetHTTPClient(cmd.tlsConfig), fb_http.WithClientRetryPeriod(cmd.RetryPeriod))
 	rc, err := client.ShardReader(ctx, indexName, shard)
 	if err != nil {
 		return fmt.Errorf("fetching shard reader: %w", err)
diff --git a/ctl/restore.go b/ctl/restore.go
index cf99619cb..589eb8138 100644
--- a/ctl/restore.go
+++ b/ctl/restore.go
@@ -5,7 +5,6 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -21,6 +20,7 @@ import (
 	fb_http "github.com/molecula/featurebase/v2/http"
 	"github.com/molecula/featurebase/v2/server"
 	"github.com/molecula/featurebase/v2/topology"
+	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -37,6 +37,9 @@ type RestoreCommand struct {
 	// Amount of time after first failed request to continue retrying.
 	RetryPeriod time.Duration `json:"retry-period"`
 
+	// Host:port on which to listen for pprof.
+	Pprof string `json:"pprof"`
+
 	// Reusable client.
 	client pilosa.InternalClient
 
@@ -51,12 +54,18 @@ func NewRestoreCommand(stdin io.Reader, stdout, stderr io.Writer) *RestoreComman
 		CmdIO:       pilosa.NewCmdIO(stdin, stdout, stderr),
 		RetryPeriod: time.Second * 30,
 		Concurrency: 1,
+		Pprof:       "localhost:43809",
 	}
 }
 
 // Run executes the restore.
 func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
 	logger := cmd.Logger()
+	close, err := startProfilingServer(cmd.Pprof, logger)
+	if err != nil {
+		return errors.Wrap(err, "starting profiling server")
+	}
+	defer close()
 
 	// Validate arguments.
 	if cmd.Path == "" {
diff --git a/ctl/util.go b/ctl/util.go
new file mode 100644
index 000000000..60ac082c9
--- /dev/null
+++ b/ctl/util.go
@@ -0,0 +1,56 @@
+package ctl
+
+import (
+	"context"
+	"net"
+	"net/http"
+	"net/http/pprof"
+	"runtime"
+	"time"
+
+	"github.com/felixge/fgprof"
+	"github.com/molecula/featurebase/v2/logger"
+	"github.com/pkg/errors"
+)
+
+// startProfilingServer starts a server which handles /debug/pprof and
+// /debug/fgprof for use in utilities we might want to profile but
+// wouldn't otherwise be running an http server. Caller should call
+// the returned close function before exiting to release resources.
+func startProfilingServer(addr string, logger logger.Logger) (close func() error, err error) {
+	if addr == "" {
+		return func() error { return nil }, nil
+	}
+
+	sm := http.NewServeMux()
+	sm.Handle("/debug/fgprof", fgprof.Handler())
+	sm.HandleFunc("/debug/pprof/", pprof.Index)
+	sm.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
+	sm.HandleFunc("/debug/pprof/profile", pprof.Profile)
+	sm.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
+	sm.HandleFunc("/debug/pprof/trace", pprof.Trace)
+	s := &http.Server{
+		Addr:    addr,
+		Handler: sm,
+	}
+	runtime.SetBlockProfileRate(10000000) // 1 sample per 10 ms
+	runtime.SetMutexProfileFraction(100)  // 1% sampling
+	ln, err := net.Listen("tcp", addr)
+	if err != nil {
+		return nil, err
+	}
+	go func() {
+		logger.Printf("Listening for /debug/pprof/ and /debug/fgprof on '%s'", addr)
+		logger.Printf("%v", s.Serve(ln))
+	}()
+
+	return func() error {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+		defer cancel()
+		err := s.Shutdown(ctx)
+		if err != nil {
+			return errors.Wrap(err, "shutting down profiling server")
+		}
+		return s.Close()
+	}, nil
+}
diff --git a/http/handler.go b/http/handler.go
index 9e626babf..958f29662 100644
--- a/http/handler.go
+++ b/http/handler.go
@@ -3348,7 +3348,7 @@ func (h *Handler) handlePostRestore(w http.ResponseWriter, r *http.Request) {
 	//validate shard for this node
 	err = h.api.RestoreShard(ctx, indexName, shard, r.Body)
 	if err != nil {
-		http.Error(w, fmt.Sprintf("failed to restore shared %v %v err:%v", indexName, shard, err), http.StatusBadRequest)
+		http.Error(w, fmt.Sprintf("failed to restore shard %v %v err:%v", indexName, shard, err), http.StatusBadRequest)
 		return
 	}
 

From 640ba45129f1df002a2df378fda4ecda50c6aa06 Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Wed, 22 Dec 2021 10:48:10 -0600
Subject: [PATCH 09/23] use retryableHTTP in client, fix memory usage of
 restore

instead of awkwardly reading an entire file into a buffer, we use
retryablehttp's reader func to open the file fresh if we need to
retry, so a small fixed-size buffer can be used internally for copying
the contents onto the network.
---
 client.go          |  14 +++++--
 cmd/slurp/slurp.go |  12 ++++--
 ctl/restore.go     |  22 ++++------
 http/client.go     | 101 ++++++++++++++++++++++-----------------------
 4 files changed, 76 insertions(+), 73 deletions(-)

diff --git a/client.go b/client.go
index fe91eb124..75741fcd3 100644
--- a/client.go
+++ b/client.go
@@ -80,8 +80,14 @@ type InternalClient interface {
 	GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[string]NodeUsage, error)
 	GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error)
 
-	ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error
-	ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error
+	// ImportFieldKeys and ImportIndexKeys are mainly used when
+	// restoring a backup. They take a readerFunc which returns a
+	// reader rather than taking an io.Reader directly to allow for
+	// efficient retries (rather than reading the entire request body
+	// into a buffer and reusing it). Reader returned from the func
+	// must be properly closed by the implementation.
+	ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, readerFunc func() (io.Reader, error)) error
+	ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, readerFunc func() (io.Reader, error)) error
 
 	// SetInternalAPI tells the client the API it should use for internal/loopback ops
 	// where applicable.
@@ -277,11 +283,11 @@ func (n nopInternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map
 func (n nopInternalClient) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error) {
 	return nil, nil
 }
-func (c nopInternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error {
+func (c nopInternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, readerFunc func() (io.Reader, error)) error {
 	return nil
 }
 
-func (c nopInternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error {
+func (c nopInternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, readerFunc func() (io.Reader, error)) error {
 	return nil
 }
 
diff --git a/cmd/slurp/slurp.go b/cmd/slurp/slurp.go
index 446db53b1..800a5a773 100644
--- a/cmd/slurp/slurp.go
+++ b/cmd/slurp/slurp.go
@@ -92,8 +92,10 @@ func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error {
 
 			byteData, err := ioutil.ReadAll(tr)
 			vprint.PanicOn(err)
-			br := bytes.NewReader(byteData)
-			err = r.client.ImportFieldKeys(context.Background(), uri, index, fieldName, false, br)
+			readerFunc := func() (io.Reader, error) {
+				return bytes.NewReader(byteData), nil
+			}
+			err = r.client.ImportFieldKeys(context.Background(), uri, index, fieldName, false, readerFunc)
 			if err != nil {
 				return err
 			}
@@ -106,9 +108,11 @@ func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error {
 			}
 			byteData, err := ioutil.ReadAll(tr)
 			vprint.PanicOn(err)
+			readerFunc := func() (io.Reader, error) {
+				return bytes.NewReader(byteData), nil
+			}
 
-			br := bytes.NewReader(byteData)
-			err = r.client.ImportIndexKeys(context.Background(), uri, index, int(partition), false, br)
+			err = r.client.ImportIndexKeys(context.Background(), uri, index, int(partition), false, readerFunc)
 			if err != nil {
 				return err
 			}
diff --git a/ctl/restore.go b/ctl/restore.go
index 589eb8138..d0c0b7ec7 100644
--- a/ctl/restore.go
+++ b/ctl/restore.go
@@ -178,7 +178,7 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.
 	return err
 }
 
-func RetryWith400(ctx context.Context, resp *http.Response, err error) (bool, error) {
+func retryWith400(ctx context.Context, resp *http.Response, err error) (bool, error) {
 	if resp != nil && resp.StatusCode > 400 { // we have some dumb status codes
 		return true, nil
 	}
@@ -202,7 +202,7 @@ func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology
 
 	client := retryablehttp.NewClient()
 	client.RetryWaitMax = cmd.RetryPeriod
-	client.CheckRetry = RetryWith400
+	client.CheckRetry = retryWith400
 	_, err = client.Post(url, "application/octet-stream", f)
 	return err
 }
@@ -281,7 +281,7 @@ func (cmd *RestoreCommand) restoreShard(ctx context.Context, filename string) er
 
 		client := retryablehttp.NewClient()
 		client.RetryWaitMax = cmd.RetryPeriod
-		client.CheckRetry = RetryWith400
+		client.CheckRetry = retryWith400
 		resp, err := client.Do(req)
 		if err != nil {
 			return err
@@ -349,13 +349,11 @@ func (cmd *RestoreCommand) restoreIndexTranslationFile(ctx context.Context, file
 
 	for _, node := range nodes {
 		if err := func() error {
-			f, err := os.Open(filename)
-			if err != nil {
-				return err
+			readerFunc := func() (io.Reader, error) {
+				return os.Open(filename) // gets used as an HTTP request body and closed by http library
 			}
-			defer f.Close()
 
-			return cmd.client.ImportIndexKeys(ctx, &node.URI, indexName, partitionID, false, f)
+			return cmd.client.ImportIndexKeys(ctx, &node.URI, indexName, partitionID, false, readerFunc)
 		}(); err != nil {
 			return err
 		}
@@ -410,13 +408,11 @@ func (cmd *RestoreCommand) restoreFieldTranslationFile(ctx context.Context, node
 
 	for _, node := range nodes {
 		if err := func() error {
-			f, err := os.Open(filename)
-			if err != nil {
-				return err
+			readerFunc := func() (io.Reader, error) {
+				return os.Open(filename)
 			}
-			defer f.Close()
 
-			return cmd.client.ImportFieldKeys(ctx, &node.URI, indexName, fieldName, false, f)
+			return cmd.client.ImportFieldKeys(ctx, &node.URI, indexName, fieldName, false, readerFunc)
 		}(); err != nil {
 			return err
 		}
diff --git a/http/client.go b/http/client.go
index c32dd86ae..a3466adac 100644
--- a/http/client.go
+++ b/http/client.go
@@ -18,6 +18,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/hashicorp/go-retryablehttp"
 	pilosa "github.com/molecula/featurebase/v2"
 	"github.com/molecula/featurebase/v2/encoding/proto"
 	"github.com/molecula/featurebase/v2/ingest"
@@ -35,10 +36,9 @@ type InternalClient struct {
 
 	log logger.Logger
 
-	retryPeriod time.Duration
-
 	// The client to use for HTTP communication.
-	httpClient *http.Client
+	httpClient      *http.Client
+	retryableClient *retryablehttp.Client
 	// the local node's API, used for operations that we can short-circuit that way
 	api *pilosa.API
 }
@@ -64,9 +64,13 @@ type InternalClientOption func(c *InternalClient)
 
 // WithClientRetryPeriod is the max amount of total time the client will
 // retry failed requests using exponential backoff.
-func WithClientRetryPeriod(period time.Duration) InternalClientOption {
+func WithClientRetryPeriod(waitMax time.Duration) InternalClientOption {
 	return func(c *InternalClient) {
-		c.retryPeriod = period
+		c.retryableClient = &retryablehttp.Client{
+			HTTPClient:   c.httpClient,
+			RetryWaitMax: waitMax,
+			CheckRetry:   retryWith400Policy,
+		}
 	}
 }
 
@@ -76,6 +80,21 @@ func WithClientLogger(log logger.Logger) InternalClientOption {
 	}
 }
 
+func noRetryPolicy(ctx context.Context, resp *http.Response, err error) (bool, error) {
+	return false, nil
+}
+
+// retryWith400Policy wraps retryablehttp's default retry policy to
+// also retry on 4XX errors which *should* be client errors and
+// therefore useless to retry, but we have some incorrect status codes.
+// TODO: fix the incorrect status codes so we can get rid of this.
+func retryWith400Policy(ctx context.Context, resp *http.Response, err error) (bool, error) {
+	if resp != nil && resp.StatusCode > 400 {
+		return true, nil
+	}
+	return retryablehttp.DefaultRetryPolicy(ctx, resp, err)
+}
+
 func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, opts ...InternalClientOption) *InternalClient {
 	ic := &InternalClient{
 		defaultURI: defaultURI,
@@ -87,6 +106,13 @@ func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, o
 	for _, opt := range opts {
 		opt(ic)
 	}
+
+	if ic.retryableClient == nil {
+		ic.retryableClient = &retryablehttp.Client{
+			HTTPClient: ic.httpClient,
+			CheckRetry: noRetryPolicy,
+		}
+	}
 	return ic
 }
 
@@ -1753,57 +1779,28 @@ func (n nopCloser) Close() error {
 	return nil
 }
 
-func (c *InternalClient) doWithRetry(req *http.Request) (*http.Response, error) {
-	sleepDuration := time.Second
-	var bod []byte
-	var err error
-	if req.Body != nil {
-		bod, err = ioutil.ReadAll(req.Body)
-		if err != nil {
-			return nil, errors.Wrap(err, "reading body")
-		}
-		req.Body = nopCloser{bytes.NewReader(bod)}
-	}
-	resp, err := c.httpClient.Do(req)
-	// start timer after first request, so if retryPeriod > 0 we
-	// pretty much always do at least one retry
-	start := time.Now()
-	for ; err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300; resp, err = c.httpClient.Do(req) {
-		if req.Body != nil {
-			req.Body = nopCloser{bytes.NewReader(bod)} // can't seek due to races with http lib internals
-		}
-		if time.Since(start) > c.retryPeriod {
-			break
-		}
-		if err != nil {
-			c.log.Printf("retrying request due to error: '%v'", err)
-		} else {
-			if bod, readErr := ioutil.ReadAll(resp.Body); readErr != nil {
-				c.log.Printf("retrying request due to status: %d, error reading body: '%v', body: '%s'", resp.StatusCode, readErr, bod)
-			} else {
-				c.log.Printf("retrying request due to status: %d, body: '%s'", resp.StatusCode, bod)
-			}
-		}
-
-		time.Sleep(sleepDuration)
-		sleepDuration *= 2
-	}
-	return resp, err
-}
-
 // executeRequest executes the given request and checks the Response. For
 // responses with non-2XX status, the body is read and closed, and an error is
 // returned. If the error is nil, the caller must ensure that the response body
 // is closed.
 func (c *InternalClient) executeRequest(req *http.Request, opts ...executeRequestOption) (*http.Response, error) {
+	return c.executeRetryableRequest(&retryablehttp.Request{Request: req}, opts...)
+}
+
+func (c *InternalClient) executeRetryableRequest(req *retryablehttp.Request, opts ...executeRequestOption) (*http.Response, error) {
+	tracing.GlobalTracer.InjectHTTPHeaders(req.Request)
+	req.Close = false
 	eo := &executeOpts{}
 	for _, opt := range opts {
 		opt(eo)
 	}
 
-	tracing.GlobalTracer.InjectHTTPHeaders(req)
-	req.Close = false
-	resp, err := c.doWithRetry(req)
+	resp, err := c.retryableClient.Do(req)
+
+	return c.handleResponse(req.Request, eo, resp, err)
+}
+
+func (c *InternalClient) handleResponse(req *http.Request, eo *executeOpts, resp *http.Response, err error) (*http.Response, error) {
 	if err != nil {
 		if resp != nil {
 			resp.Body.Close()
@@ -2083,7 +2080,7 @@ func (c *InternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context,
 
 	return resp.Body, nil
 }
-func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error {
+func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, readerFunc func() (io.Reader, error)) error {
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportIndexKeys")
 	defer span.Finish()
 
@@ -2100,14 +2097,14 @@ func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, ind
 	url := fmt.Sprintf("%s/internal/translate/index/%s/%d", uri, index, partitionID)
 
 	// Generate HTTP request.
-	httpReq, err := http.NewRequest("POST", url, rddbdata)
+	httpReq, err := retryablehttp.NewRequest("POST", url, readerFunc)
 	if err != nil {
 		return errors.Wrap(err, "creating request")
 	}
 	httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
 
 	// Execute request against the host.
-	resp, err := c.executeRequest(httpReq.WithContext(ctx))
+	resp, err := c.executeRetryableRequest(httpReq.WithContext(ctx))
 	if err != nil {
 		return err
 	}
@@ -2115,7 +2112,7 @@ func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, ind
 	return nil
 }
 
-func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error {
+func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, readerFunc func() (io.Reader, error)) error {
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportFieldKeys")
 	defer span.Finish()
 
@@ -2132,14 +2129,14 @@ func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, ind
 	url := fmt.Sprintf("%s/internal/translate/field/%s/%s", uri, index, field)
 
 	// Generate HTTP request.
-	httpReq, err := http.NewRequest("POST", url, rddbdata)
+	httpReq, err := retryablehttp.NewRequest("POST", url, readerFunc)
 	if err != nil {
 		return errors.Wrap(err, "creating request")
 	}
 	httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
 
 	// Execute request against the host.
-	resp, err := c.executeRequest(httpReq.WithContext(ctx))
+	resp, err := c.executeRetryableRequest(httpReq.WithContext(ctx))
 	if err != nil {
 		return err
 	}

From ea59f14d5025138a91b6bfff154c8f9fa83dd43a Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Wed, 22 Dec 2021 11:21:11 -0600
Subject: [PATCH 10/23] must use retryablehttp.NewClient to get defaults

otherwise it won't actually retry :(
---
 http/client.go | 20 +++++++++++---------
 1 file changed, 11 insertions(+), 9 deletions(-)

diff --git a/http/client.go b/http/client.go
index a3466adac..7f7374d4d 100644
--- a/http/client.go
+++ b/http/client.go
@@ -66,11 +66,12 @@ type InternalClientOption func(c *InternalClient)
 // retry failed requests using exponential backoff.
 func WithClientRetryPeriod(waitMax time.Duration) InternalClientOption {
 	return func(c *InternalClient) {
-		c.retryableClient = &retryablehttp.Client{
-			HTTPClient:   c.httpClient,
-			RetryWaitMax: waitMax,
-			CheckRetry:   retryWith400Policy,
-		}
+		fmt.Println("client w/ retry policy", waitMax)
+		rc := retryablehttp.NewClient()
+		rc.HTTPClient = c.httpClient
+		rc.RetryWaitMax = waitMax
+		rc.CheckRetry = retryWith400Policy
+		c.retryableClient = rc
 	}
 }
 
@@ -108,10 +109,11 @@ func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, o
 	}
 
 	if ic.retryableClient == nil {
-		ic.retryableClient = &retryablehttp.Client{
-			HTTPClient: ic.httpClient,
-			CheckRetry: noRetryPolicy,
-		}
+		fmt.Println("no retry policy")
+		rc := retryablehttp.NewClient()
+		rc.HTTPClient = ic.httpClient
+		rc.CheckRetry = noRetryPolicy
+		ic.retryableClient = rc
 	}
 	return ic
 }

From 295fab4892d81336b71c49a6d0a4ec6e790e6e7a Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Wed, 22 Dec 2021 12:21:11 -0600
Subject: [PATCH 11/23] retry on >= 400, not just greater. good catch

---
 ctl/restore.go | 2 +-
 http/client.go | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/ctl/restore.go b/ctl/restore.go
index d0c0b7ec7..f7d669f09 100644
--- a/ctl/restore.go
+++ b/ctl/restore.go
@@ -179,7 +179,7 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.
 }
 
 func retryWith400(ctx context.Context, resp *http.Response, err error) (bool, error) {
-	if resp != nil && resp.StatusCode > 400 { // we have some dumb status codes
+	if resp != nil && resp.StatusCode >= 400 { // we have some dumb status codes
 		return true, nil
 	}
 	return retryablehttp.DefaultRetryPolicy(ctx, resp, err)
diff --git a/http/client.go b/http/client.go
index 7f7374d4d..222499794 100644
--- a/http/client.go
+++ b/http/client.go
@@ -90,7 +90,7 @@ func noRetryPolicy(ctx context.Context, resp *http.Response, err error) (bool, e
 // therefore useless to retry, but we have some incorrect status codes.
 // TODO: fix the incorrect status codes so we can get rid of this.
 func retryWith400Policy(ctx context.Context, resp *http.Response, err error) (bool, error) {
-	if resp != nil && resp.StatusCode > 400 {
+	if resp != nil && resp.StatusCode >= 400 {
 		return true, nil
 	}
 	return retryablehttp.DefaultRetryPolicy(ctx, resp, err)

From 9367a626095ecfb10d702f8abdab0415b807d9c1 Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Mon, 27 Dec 2021 09:34:43 -0700
Subject: [PATCH 12/23] Add /debug/rbf endpoint for debugging

---
 api.go          | 16 ++++++++++++++++
 api_test.go     | 22 ++++++++++++++++++++++
 http/handler.go | 15 +++++++++++++++
 rbf/db.go       | 17 +++++++++++++++++
 rbf/db_test.go  | 15 +++++++++++++++
 rbf/tx.go       | 17 +++++++++++++++++
 6 files changed, 102 insertions(+)

diff --git a/api.go b/api.go
index 47558678d..628feb16d 100644
--- a/api.go
+++ b/api.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/molecula/featurebase/v2/disco"
 	"github.com/molecula/featurebase/v2/ingest"
+	"github.com/molecula/featurebase/v2/rbf"
 
 	//"github.com/molecula/featurebase/v2/pg"
 	"github.com/molecula/featurebase/v2/pql"
@@ -3156,6 +3157,21 @@ func (api *API) Plan(ctx context.Context, q string) (*Stmt, error) {
 	return api.server.PlanSQL(ctx, q)
 }
 
+func (api *API) RBFDebugInfo() map[string]*rbf.DebugInfo {
+	infos := make(map[string]*rbf.DebugInfo)
+
+	for key, dbShard := range api.holder.Txf().dbPerShard.Flatmap {
+		wrapper, ok := dbShard.W.(*RbfDBWrapper)
+		if !ok {
+			continue
+		}
+
+		skey := fmt.Sprintf("%s/%d", key.index, key.shard)
+		infos[skey] = wrapper.db.DebugInfo()
+	}
+	return infos
+}
+
 type serverInfo struct {
 	ShardWidth       uint64 `json:"shardWidth"`
 	ReplicaN         int    `json:"replicaN"`
diff --git a/api_test.go b/api_test.go
index 9ffdea841..d42aca667 100644
--- a/api_test.go
+++ b/api_test.go
@@ -1415,3 +1415,25 @@ func TestVariousApiTranslateCalls(t *testing.T) {
 		*/
 	}
 }
+
+func TestAPI_RBFDebugInfo(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c := test.MustRunCluster(t, 1,
+		[]server.CommandOption{
+			server.OptCommandServerOptions(
+				pilosa.OptServerNodeID("node0"),
+				pilosa.OptServerClusterHasher(&offsetModHasher{}),
+				pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
+			)},
+	)
+	defer c.Close()
+
+	coord := c.GetPrimary()
+
+	if _, err := coord.API.CreateIndex(ctx, "i", pilosa.IndexOptions{}); err != nil {
+		t.Fatal(err)
+	} else if infos := coord.API.RBFDebugInfo(); infos == nil {
+		t.Fatal("expected info")
+	}
+}
diff --git a/http/handler.go b/http/handler.go
index 958f29662..05b65aa58 100644
--- a/http/handler.go
+++ b/http/handler.go
@@ -441,6 +441,9 @@ func newRouter(handler *Handler) http.Handler {
 	router.HandleFunc("/internal/idalloc/data", handler.handleIDAllocData).Methods("GET").Name("IDAllocData")
 
 	router.HandleFunc("/internal/restore/{index}/{shardID}", handler.handlePostRestore).Methods("POST").Name("Restore")
+
+	router.HandleFunc("/internal/debug/rbf", handler.handleGetInternalDebugRBFJSON).Methods("GET").Name("GetInternalDebugRBFJSON")
+
 	// endpoints for collecting cpu profiles from a chosen begin point to
 	// when the client wants to stop. Used for profiling imports that
 	// could be long or short.
@@ -2064,6 +2067,18 @@ func validateProtobufHeader(r *http.Request) (error string, code int) {
 	return
 }
 
+// handleGetInternalDebugRBFJSON handles /internal/debug/rbf requests.
+func (h *Handler) handleGetInternalDebugRBFJSON(w http.ResponseWriter, r *http.Request) {
+	buf, err := json.MarshalIndent(h.api.RBFDebugInfo(), "", "  ")
+	if err != nil {
+		http.Error(w, "marshal json: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	w.Write(buf)
+}
+
 // handleGetMetricsJSON handles /metrics.json requests, translating text metrics results to more consumable JSON.
 func (h *Handler) handleGetMetricsJSON(w http.ResponseWriter, r *http.Request) {
 	if !validHeaderAcceptJSON(r.Header) {
diff --git a/rbf/db.go b/rbf/db.go
index 65dd4fbea..6e4955f94 100644
--- a/rbf/db.go
+++ b/rbf/db.go
@@ -7,6 +7,8 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"runtime/debug"
+	"sort"
 	"sync"
 	"syscall"
 
@@ -637,6 +639,7 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) {
 		pageMap:     db.pageMap,
 		walPageN:    db.walPageN,
 		writable:    writable,
+		stack:       debug.Stack(), // DEBUG
 
 		DeleteEmptyContainer: true,
 	}
@@ -815,6 +818,20 @@ func (db *DB) getCursor(tx *Tx) *Cursor {
 	return c
 }
 
+func (db *DB) DebugInfo() *DebugInfo {
+	info := &DebugInfo{Path: db.Path}
+	for tx := range db.txs {
+		info.Txs = append(info.Txs, tx.DebugInfo())
+	}
+	sort.Slice(info.Txs, func(i, j int) bool { return info.Txs[i].Ptr < info.Txs[j].Ptr })
+	return info
+}
+
+type DebugInfo struct {
+	Path string         `json:"path"`
+	Txs  []*TxDebugInfo `json:"txs"`
+}
+
 // Shared pool for in-memory database pages.
 // These are used before being flushed to disk.
 var pagePool = &sync.Pool{
diff --git a/rbf/db_test.go b/rbf/db_test.go
index c0eb3b3c7..8bae550bb 100644
--- a/rbf/db_test.go
+++ b/rbf/db_test.go
@@ -339,6 +339,21 @@ func TestDB_MultiTx(t *testing.T) {
 	}
 }
 
+func TestDB_DebugInfo(t *testing.T) {
+	db := MustOpenDB(t)
+	defer MustCloseDB(t, db)
+
+	tx := MustBegin(t, db, true)
+	defer tx.Rollback()
+
+	info := db.DebugInfo()
+	if got, want := info.Path, db.Path; got != want {
+		t.Fatalf("Path=%q, want %q", got, want)
+	} else if got, want := len(info.Txs), 1; got != want {
+		t.Fatalf("len(Txs)=%d, want %d", got, want)
+	}
+}
+
 // premake pool of random values
 const randPool = (1 << 18)
 
diff --git a/rbf/tx.go b/rbf/tx.go
index 5c6c7f7c6..5d731fe4f 100644
--- a/rbf/tx.go
+++ b/rbf/tx.go
@@ -65,6 +65,9 @@ type Tx struct {
 	// manages to trigger a *deallocation* (which I don't think should be
 	// happening), we'll process that one after the current list is processed.
 	pendingFreelistAdds []uint32
+
+	// DEBUG
+	stack []byte
 }
 
 func (tx *Tx) DBPath() string {
@@ -2042,6 +2045,20 @@ func (tx *Tx) GetSortedFieldViewList() (fvs []txkey.FieldView, _ error) {
 	return
 }
 
+func (tx *Tx) DebugInfo() *TxDebugInfo {
+	return &TxDebugInfo{
+		Ptr:      fmt.Sprintf("%p", tx),
+		Writable: tx.writable,
+		Stack:    string(tx.stack),
+	}
+}
+
+type TxDebugInfo struct {
+	Ptr      string `json:"ptr"`
+	Writable bool   `json:"writable"`
+	Stack    string `json:"stack,omitempty"`
+}
+
 // SnapshotReader returns a reader that provides a snapshot for the current database state.
 func (tx *Tx) SnapshotReader() (io.Reader, error) {
 	if tx.db == nil {

From 6638fa17eeb36ca4ffcbcf5c4fb0ab053291747d Mon Sep 17 00:00:00 2001
From: Travis <travis@pilosa.com>
Date: Mon, 27 Dec 2021 10:22:04 -0600
Subject: [PATCH 13/23] Expose `etcd.dir` configuration option

The goal is to allow a user to separate FeatureBase and etcd I/O.
---
 ctl/server.go                            | 2 +-
 internal/clustertests/docker-compose.yml | 3 +++
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/ctl/server.go b/ctl/server.go
index 9867c35b9..086506f54 100644
--- a/ctl/server.go
+++ b/ctl/server.go
@@ -45,7 +45,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
 
 	// Etcd
 	// Etcd.Name used Config.Name for its value.
-	// Etcd.Dir defaults to a directory under the pilosa data directory.
+	flags.StringVar(&srv.Config.Etcd.Dir, "etcd.dir", srv.Config.Etcd.Dir, "Directory to store etcd data files. If not provided, a directory will be created under the main data-dir directory.")
 	// Etcd.ClusterName uses Cluster.Name for its value
 	flags.StringVar(&srv.Config.Etcd.LClientURL, "etcd.listen-client-address", srv.Config.Etcd.LClientURL, "Listen client address.")
 	flags.StringVar(&srv.Config.Etcd.AClientURL, "etcd.advertise-client-address", srv.Config.Etcd.AClientURL, "Advertise client address. If not provided, uses the listen client address.")
diff --git a/internal/clustertests/docker-compose.yml b/internal/clustertests/docker-compose.yml
index 4192be6f8..46143a3f6 100644
--- a/internal/clustertests/docker-compose.yml
+++ b/internal/clustertests/docker-compose.yml
@@ -9,6 +9,7 @@ services:
       - "33455:10101"
     environment:
       - PILOSA_NAME=pilosa1
+      - PILOSA_ETCD_DIR=/root/.etcd
       - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
       - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa1:10201
       - PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301
@@ -28,6 +29,7 @@ services:
       - "33456:10101"
     environment:
       - PILOSA_NAME=pilosa2
+      - PILOSA_ETCD_DIR=/root/.etcd
       - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
       - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa2:10201
       - PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301
@@ -47,6 +49,7 @@ services:
       - "33457:10101"
     environment:
       - PILOSA_NAME=pilosa3
+      - PILOSA_ETCD_DIR=/root/.etcd
       - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
       - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa3:10201
       - PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301

From 310584b0d8d69c9e3e5654bc2bf950109a90526a Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Mon, 27 Dec 2021 13:30:57 -0700
Subject: [PATCH 14/23] Add job & worker metrics

---
 executor.go | 20 ++++++++++++++++++--
 server.go   |  4 ++++
 2 files changed, 22 insertions(+), 2 deletions(-)

diff --git a/executor.go b/executor.go
index 5e03e1a72..f0adc7bc7 100644
--- a/executor.go
+++ b/executor.go
@@ -179,11 +179,18 @@ func newExecutor(opts ...executorOption) *executor {
 
 func (e *executor) addWorker() {
 	e.workersWG.Add(1)
-	atomic.AddInt64(&e.currentWorkers, 1)
+	n := atomic.AddInt64(&e.currentWorkers, 1)
+	if e.Holder != nil {
+		e.Holder.Stats.Gauge("worker_total", float64(n), 0)
+	}
+
 	go func() {
 		defer e.workersWG.Done()
 		e.worker(e.work)
-		atomic.AddInt64(&e.currentWorkers, -1)
+		n := atomic.AddInt64(&e.currentWorkers, -1)
+		if e.Holder != nil {
+			e.Holder.Stats.Gauge("worker_total", float64(n), 0)
+		}
 	}()
 }
 
@@ -204,6 +211,14 @@ func (e *executor) Close() error {
 	return nil
 }
 
+// InitStats initializes stats counters. Must be called after Holder set.
+func (e *executor) InitStats() {
+	if e.Holder != nil {
+		e.Holder.Stats.Count("job_total", 0, 0)
+		e.Holder.Stats.Gauge("worker_total", float64(atomic.LoadInt64(&e.currentWorkers)), 0)
+	}
+}
+
 // Execute executes a PQL query.
 func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) (QueryResponse, error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.Execute")
@@ -5989,6 +6004,7 @@ type job struct {
 func (e *executor) worker(work chan job) {
 	for j := range work {
 		atomic.AddUint64(&e.workCounter, 1)
+		e.Holder.Stats.Count("job_total", 1, 0)
 		if j.idleHands {
 			return
 		}
diff --git a/server.go b/server.go
index 0c74a2fa5..a5d363e80 100644
--- a/server.go
+++ b/server.go
@@ -506,6 +506,10 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 	s.holder.schemator = s.schemator
 	s.holder.sharder = s.sharder
 	s.holder.serializer = s.serializer
+
+	// Initial stats must be invoked after the executor obtains reference to the holder.
+	s.executor.InitStats()
+
 	return s, nil
 }
 

From bb39b05d0572e42d4b23090441a204e55b969aae Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Mon, 27 Dec 2021 13:02:22 -0600
Subject: [PATCH 15/23] remove leftover fmt.Println

---
 http/client.go | 1 -
 1 file changed, 1 deletion(-)

diff --git a/http/client.go b/http/client.go
index 222499794..1dd2da51a 100644
--- a/http/client.go
+++ b/http/client.go
@@ -109,7 +109,6 @@ func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, o
 	}
 
 	if ic.retryableClient == nil {
-		fmt.Println("no retry policy")
 		rc := retryablehttp.NewClient()
 		rc.HTTPClient = ic.httpClient
 		rc.CheckRetry = noRetryPolicy

From fe54cbf8ae31f64f269157e483bb7df8abbb4b21 Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Mon, 27 Dec 2021 14:27:59 -0600
Subject: [PATCH 16/23] remove other print and tweak backup test timings

---
 http/client.go                        | 1 -
 internal/clustertests/cluster_test.go | 2 +-
 2 files changed, 1 insertion(+), 2 deletions(-)

diff --git a/http/client.go b/http/client.go
index 1dd2da51a..e343c65c9 100644
--- a/http/client.go
+++ b/http/client.go
@@ -66,7 +66,6 @@ type InternalClientOption func(c *InternalClient)
 // retry failed requests using exponential backoff.
 func WithClientRetryPeriod(waitMax time.Duration) InternalClientOption {
 	return func(c *InternalClient) {
-		fmt.Println("client w/ retry policy", waitMax)
 		rc := retryablehttp.NewClient()
 		rc.HTTPClient = c.httpClient
 		rc.RetryWaitMax = waitMax
diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go
index af89abc2a..d3009dbda 100644
--- a/internal/clustertests/cluster_test.go
+++ b/internal/clustertests/cluster_test.go
@@ -153,7 +153,7 @@ func TestClusterStuff(t *testing.T) {
 			t.Fatalf("sending stop command: %v", err)
 		}
 		if backupCmd, err = startCmd(
-			"featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest2"), "--retry-period=0.5s"); err != nil {
+			"featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest2"), "--retry-period=50ms"); err != nil {
 			t.Fatalf("sending second backup command: %v", err)
 		}
 		time.Sleep(time.Millisecond * 5) // want the backup to get started, then fail

From 1a8c10d5f3adebd45971a2ab770f0dbaf8778538 Mon Sep 17 00:00:00 2001
From: Matthew Jaffee <jaffee@pilosa.com>
Date: Tue, 28 Dec 2021 10:31:05 -0600
Subject: [PATCH 17/23] fix backup fail test so it actually fails

A few things were going wrong here.

First, we take a "RetryPeriod" option on backup and restore which is
meant to be roughly the total amount of time we spend retrying any
given request before failing. However we were incorrectly passing that
as the RetryMaxWait which is the maximum amount of time to sleep
between any two attempts. We now do some fuzzy math to figure out
approximately how many attempts we should make given a minimum sleep
of 100ms and the fact that we double the sleep time every attempt.

Second, during the backup test, if a host was totally stopped when we
started the request, it would fail immediately and then retry, but if
the host was stopped during the request (after DNS had resolved), then
the request would wait for the DialTimeout which we default to 30s, so
turning off the cluster for 5 seconds and turning it back on resulted
in the backup completing rather than failing. Because of this, we
change the commandClient to have a default dial timeout of 1 second.

I was tempted to change the global default to 1s which I think would
be fine, but didn't want to break anything too badly.
---
 ctl/common.go                         | 15 +++++++++++-
 ctl/restore.go                        | 33 ++++++++++++++++++++-------
 http/client.go                        | 16 +++++++++++--
 http/handler.go                       | 24 ++++++++++++-------
 internal/clustertests/cluster_test.go | 11 +++++----
 5 files changed, 75 insertions(+), 24 deletions(-)

diff --git a/ctl/common.go b/ctl/common.go
index 7ba2df2f7..558cdece3 100644
--- a/ctl/common.go
+++ b/ctl/common.go
@@ -2,6 +2,11 @@
 package ctl
 
 import (
+	"net"
+	"time"
+
+	gohttp "net/http"
+
 	"github.com/molecula/featurebase/v2/http"
 	"github.com/molecula/featurebase/v2/logger"
 	"github.com/molecula/featurebase/v2/server"
@@ -25,6 +30,14 @@ func SetTLSConfig(flags *pflag.FlagSet, prefix string, certificatePath *string,
 	flags.BoolVarP(enableClientVerification, prefix+"tls.enable-client-verification", "", false, "Enable TLS certificate client verification for incoming connections")
 }
 
+// default dial timeout is 30s for some reason which makes testing
+// failures/retries really awkward. I don't think we need it that
+// high, so I set it to 1s here... let's see what happens.
+func clientOptions(client *gohttp.Client, dialer *net.Dialer) *gohttp.Client {
+	dialer.Timeout = time.Second * 1
+	return client
+}
+
 // commandClient returns a pilosa.InternalHTTPClient for the command
 func commandClient(cmd CommandWithTLSSupport, opts ...http.InternalClientOption) (*http.InternalClient, error) {
 	tls := cmd.TLSConfiguration()
@@ -32,7 +45,7 @@ func commandClient(cmd CommandWithTLSSupport, opts ...http.InternalClientOption)
 	if err != nil {
 		return nil, errors.Wrap(err, "getting tls config")
 	}
-	client, err := http.NewInternalClient(cmd.TLSHost(), http.GetHTTPClient(tlsConfig), opts...)
+	client, err := http.NewInternalClient(cmd.TLSHost(), http.GetHTTPClient(tlsConfig, clientOptions), opts...)
 	if err != nil {
 		return nil, errors.Wrap(err, "getting internal client")
 	}
diff --git a/ctl/restore.go b/ctl/restore.go
index f7d669f09..b7f1fb2dc 100644
--- a/ctl/restore.go
+++ b/ctl/restore.go
@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"math"
 	"net/http"
 	"os"
 	"path/filepath"
@@ -137,8 +138,7 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.
 	if len(existingSchema) == 0 {
 		cmd.Logger().Printf("Load Schema")
 		url := primary.URI.Path("/schema")
-		client := retryablehttp.NewClient()
-		client.RetryWaitMax = cmd.RetryPeriod
+		client := cmd.newClient()
 		_, err = client.Post(url, "application/json", f)
 	} else {
 		schema := &pilosa.Schema{}
@@ -185,6 +185,27 @@ func retryWith400(ctx context.Context, resp *http.Response, err error) (bool, er
 	return retryablehttp.DefaultRetryPolicy(ctx, resp, err)
 }
 
+// This logic is taken from featurebase/http/client.go If this logic
+// is not the same as what's there, that could be a problem. Ideally
+// all network calls from restore would go through the client and this
+// would not longer be needed.
+func (cmd *RestoreCommand) newClient() *retryablehttp.Client {
+	min := time.Millisecond * 100
+
+	// do some math to figure out how many attempts we need to get our
+	// total sleep time close to the period
+	attempts := math.Log2(float64(cmd.RetryPeriod)) - math.Log2(float64(min))
+	attempts += 0.3 // mmmm, fudge
+	if attempts < 1 {
+		attempts = 1
+	}
+	client := retryablehttp.NewClient()
+	client.RetryWaitMin = min
+	client.RetryMax = int(attempts)
+	client.CheckRetry = retryWith400
+	return client
+}
+
 func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology.Node) error {
 	logger := cmd.Logger()
 
@@ -200,9 +221,7 @@ func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology
 	logger.Printf("Load idalloc")
 	url := primary.URI.Path("/internal/idalloc/restore")
 
-	client := retryablehttp.NewClient()
-	client.RetryWaitMax = cmd.RetryPeriod
-	client.CheckRetry = retryWith400
+	client := cmd.newClient()
 	_, err = client.Post(url, "application/octet-stream", f)
 	return err
 }
@@ -279,9 +298,7 @@ func (cmd *RestoreCommand) restoreShard(ctx context.Context, filename string) er
 		req = req.WithContext(ctx)
 		req.Header.Set("Content-Type", "application/octet-stream")
 
-		client := retryablehttp.NewClient()
-		client.RetryWaitMax = cmd.RetryPeriod
-		client.CheckRetry = retryWith400
+		client := cmd.newClient()
 		resp, err := client.Do(req)
 		if err != nil {
 			return err
diff --git a/http/client.go b/http/client.go
index e343c65c9..8d3437bb5 100644
--- a/http/client.go
+++ b/http/client.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"math"
 	"math/rand"
 	"net/http"
 	"net/url"
@@ -64,11 +65,22 @@ type InternalClientOption func(c *InternalClient)
 
 // WithClientRetryPeriod is the max amount of total time the client will
 // retry failed requests using exponential backoff.
-func WithClientRetryPeriod(waitMax time.Duration) InternalClientOption {
+func WithClientRetryPeriod(period time.Duration) InternalClientOption {
+	min := time.Millisecond * 100
+
+	// do some math to figure out how many attempts we need to get our
+	// total sleep time close to the period
+	attempts := math.Log2(float64(period)) - math.Log2(float64(min))
+	attempts += 0.3 // mmmm, fudge
+	if attempts < 1 {
+		attempts = 1
+	}
+	fmt.Println("attempts: ", int(attempts))
 	return func(c *InternalClient) {
 		rc := retryablehttp.NewClient()
 		rc.HTTPClient = c.httpClient
-		rc.RetryWaitMax = waitMax
+		rc.RetryWaitMin = min
+		rc.RetryMax = int(attempts)
 		rc.CheckRetry = retryWith400Policy
 		c.retryableClient = rc
 	}
diff --git a/http/handler.go b/http/handler.go
index 05b65aa58..7ef5eda37 100644
--- a/http/handler.go
+++ b/http/handler.go
@@ -2593,14 +2593,17 @@ func (s queryValidationSpec) validate(query url.Values) error {
 	return nil
 }
 
-func GetHTTPClient(t *tls.Config) *http.Client {
+type ClientOption func(client *http.Client, dialer *net.Dialer) *http.Client
+
+func GetHTTPClient(t *tls.Config, opts ...ClientOption) *http.Client {
+	dialer := &net.Dialer{
+		Timeout:   30 * time.Second,
+		KeepAlive: 30 * time.Second,
+		DualStack: true,
+	}
 	transport := &http.Transport{
-		Proxy: http.ProxyFromEnvironment,
-		DialContext: (&net.Dialer{
-			Timeout:   30 * time.Second,
-			KeepAlive: 30 * time.Second,
-			DualStack: true,
-		}).DialContext,
+		Proxy:                 http.ProxyFromEnvironment,
+		DialContext:           dialer.DialContext,
 		MaxIdleConns:          1000,
 		MaxIdleConnsPerHost:   200,
 		IdleConnTimeout:       90 * time.Second,
@@ -2610,7 +2613,12 @@ func GetHTTPClient(t *tls.Config) *http.Client {
 	if t != nil {
 		transport.TLSClientConfig = t
 	}
-	return &http.Client{Transport: transport}
+
+	client := &http.Client{Transport: transport}
+	for _, opt := range opts {
+		client = opt(client, dialer)
+	}
+	return client
 }
 
 // handlePostImportAtomicRecord handles /import-atomic-record requests
diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go
index d3009dbda..e23d10f8c 100644
--- a/internal/clustertests/cluster_test.go
+++ b/internal/clustertests/cluster_test.go
@@ -149,22 +149,23 @@ func TestClusterStuff(t *testing.T) {
 		// now do backup with all nodes down and too short a timeout
 		// so it fails. Has be to be all 3 because the cluster has
 		// replicas=3 and the backup command will retry on replicas.
-		if err = sendCmd("docker", "stop", "clustertests_pilosa2_1"); err != nil {
-			t.Fatalf("sending stop command: %v", err)
-		}
 		if backupCmd, err = startCmd(
-			"featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest2"), "--retry-period=50ms"); err != nil {
+			"featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest2"), "--retry-period=200ms"); err != nil {
 			t.Fatalf("sending second backup command: %v", err)
 		}
-		time.Sleep(time.Millisecond * 5) // want the backup to get started, then fail
+		time.Sleep(time.Millisecond * 10) // want the backup to get started, then fail
 		if err = sendCmd("docker", "stop", "clustertests_pilosa1_1"); err != nil {
 			t.Fatalf("sending stop command: %v", err)
 		}
+		if err = sendCmd("docker", "stop", "clustertests_pilosa2_1"); err != nil {
+			t.Fatalf("sending stop command: %v", err)
+		}
 		if err = sendCmd("docker", "stop", "clustertests_pilosa3_1"); err != nil {
 			t.Fatalf("sending stop command: %v", err)
 		}
 
 		time.Sleep(time.Second * 5)
+
 		if err = sendCmd("docker", "start", "clustertests_pilosa1_1"); err != nil {
 			t.Fatalf("sending start command: %v", err)
 		}

From ffd91137e1702de482bbe9ef00fe59410d743be5 Mon Sep 17 00:00:00 2001
From: Travis <travis@pilosa.com>
Date: Tue, 28 Dec 2021 13:34:07 -0600
Subject: [PATCH 18/23] Stop blocking API called when cluster is DOWN or
 DEGRADED

This commit effectively removes the API-level validation that was
blocking certain API methods when the cluster was in a particular state
(namely DOWN and DEGRADED). The thinking is that we shouldn't be
blocking these requests at the API level, but rather should let them
pass through and allow the fact that a node is ACTUALLY down dictate the
behavior.

With this change, two tests were modified. They were previously
expecting the error message from the API validation on DOWN, but now
they check for a "shard unavailable" error, which is what gets returned
for a particular query when the cluster is in an unhealthy state.
---
 api.go                | 11 +++++-
 server/server_test.go | 86 +++++++++++++++++++++++++++++++++++++++++--
 2 files changed, 91 insertions(+), 6 deletions(-)

diff --git a/api.go b/api.go
index 628feb16d..fdee2f8f7 100644
--- a/api.go
+++ b/api.go
@@ -131,9 +131,16 @@ func (api *API) SetAPIOptions(opts ...apiOption) error {
 var validAPIMethods = map[disco.ClusterState]map[apiMethod]struct{}{
 	disco.ClusterStateStarting: methodsCommon,
 	disco.ClusterStateNormal:   appendMap(methodsCommon, methodsNormal),
-	disco.ClusterStateDegraded: appendMap(methodsCommon, methodsDegraded),
+	// Ideally, this would be just `appendMap(methodsCommon, methodsDegraded)`,
+	// but in an attempt to reduce the influence that state (determined by etcd)
+	// has on a node under load, this is set to effectively allow all requests
+	// in a DEGRADED state.
+	disco.ClusterStateDegraded: appendMap(methodsCommon, methodsNormal),
 	disco.ClusterStateResizing: appendMap(methodsCommon, methodsResizing),
-	disco.ClusterStateDown:     methodsCommon,
+	// Ideally, this would be just `methodsCommon`, but in an attempt to reduce
+	// the influence that state (determined by etcd) has on a node under load,
+	// this is set to effectively allow all requests in a DOWN state.
+	disco.ClusterStateDown: appendMap(methodsCommon, methodsNormal),
 }
 
 func appendMap(a, b map[apiMethod]struct{}) map[apiMethod]struct{} {
diff --git a/server/server_test.go b/server/server_test.go
index 014e2035e..551781ffb 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -17,7 +17,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/molecula/featurebase/v2"
+	pilosa "github.com/molecula/featurebase/v2"
 	"github.com/molecula/featurebase/v2/disco"
 	"github.com/molecula/featurebase/v2/http"
 	"github.com/molecula/featurebase/v2/pql"
@@ -26,6 +26,7 @@ import (
 	"github.com/molecula/featurebase/v2/test"
 	"github.com/molecula/featurebase/v2/testhook"
 	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -504,6 +505,30 @@ func TestClusteringNodesReplica1(t *testing.T) {
 		t.Fatalf("starting cluster: %v", err)
 	}
 
+	indexName := "idx"
+	fieldName := "fld"
+
+	// Create the schema.
+	if _, err := cluster.GetPrimary().API.CreateIndex(context.Background(), indexName, pilosa.IndexOptions{}); err != nil {
+		t.Fatalf("creating index: %v", err)
+	}
+	if _, err := cluster.GetPrimary().API.CreateField(context.Background(), indexName, fieldName); err != nil {
+		t.Fatalf("creating field: %v", err)
+	}
+
+	// Set some columns across shards to ensure that the Row query will require
+	// data from all nodes.
+	data := []string{}
+	for rowID := 1; rowID < 2; rowID++ {
+		for columnID := 1; columnID < 10; columnID++ {
+			data = append(data, fmt.Sprintf(`Set(%d, %s=%d)`, columnID*pilosa.ShardWidth, fieldName, rowID))
+		}
+	}
+	if _, err := cluster.GetPrimary().Query(t, indexName, "", strings.Join(data, "")); err != nil {
+		t.Fatalf("setting columns: %v", err)
+	}
+
+	// Shut down a node.
 	if err := cluster.GetNonPrimary().Command.Close(); err != nil {
 		t.Fatalf("closing third node: %v", err)
 	}
@@ -513,7 +538,12 @@ func TestClusteringNodesReplica1(t *testing.T) {
 	}
 
 	// confirm that cluster stops accepting queries after one node closes
-	if _, err := cluster.GetPrimary().API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state DOWN") {
+	qry := &pilosa.QueryRequest{
+		Index: "idx",
+		Query: fmt.Sprintf("Row(%s=1)", fieldName),
+	}
+
+	if _, err := cluster.GetPrimary().API.Query(context.Background(), qry); !strings.Contains(err.Error(), "shard unavailable") {
 		t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
 	}
 }
@@ -540,8 +570,34 @@ func TestClusteringNodesReplica2(t *testing.T) {
 	}
 	defer cluster.Close()
 
+	indexName := "idx"
+	fieldName := "fld"
+
 	coord, others := cluster.GetPrimary(), cluster.GetNonPrimaries()
 
+	// Create the schema.
+	if _, err := coord.API.CreateIndex(context.Background(), indexName, pilosa.IndexOptions{}); err != nil {
+		t.Fatalf("creating index: %v", err)
+	}
+	if _, err := coord.API.CreateField(context.Background(), indexName, fieldName); err != nil {
+		t.Fatalf("creating field: %v", err)
+	}
+
+	// Set some columns across shards to ensure that the Row query will require
+	// data from all nodes.
+	data := []string{}
+	cols := []uint64{}
+	for rowID := 1; rowID < 2; rowID++ {
+		for columnID := 1; columnID < 30; columnID++ {
+			col := uint64(columnID * pilosa.ShardWidth)
+			cols = append(cols, col)
+			data = append(data, fmt.Sprintf(`Set(%d, %s=%d)`, col, fieldName, rowID))
+		}
+	}
+	if _, err := coord.Query(t, indexName, "", strings.Join(data, "")); err != nil {
+		t.Fatalf("setting columns: %v", err)
+	}
+
 	if err := others[0].Close(); err != nil {
 		t.Fatalf("closing third node: %v", err)
 	}
@@ -569,8 +625,30 @@ func TestClusteringNodesReplica2(t *testing.T) {
 		t.Fatalf("after closing second server: %v", err)
 	}
 
-	if _, err := coord.API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state DOWN") {
-		t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
+	qry := &pilosa.QueryRequest{
+		Index: "idx",
+		Query: fmt.Sprintf("Row(%s=1)", fieldName),
+	}
+
+	// Because we no longer block queries when the cluster is in state DOWN,
+	// there are cases where a DOWN cluster can still respond to a query. In
+	// that case, we want the test to pass. But if the unavailable node(s) cause
+	// the query to result in an error, we check that it's the error we expect.
+	resp, err := coord.API.Query(context.Background(), qry)
+	if err != nil {
+		if !strings.Contains(err.Error(), "shard unavailable") {
+			t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
+		}
+	} else {
+		if len(resp.Results) == 0 {
+			t.Fatal("got no results")
+		}
+
+		row, ok := resp.Results[0].(*pilosa.Row)
+		if !ok {
+			t.Fatalf("expected a *pilosa.Row, but got %T", resp.Results[0])
+		}
+		require.Equal(t, row.Columns(), cols)
 	}
 }
 

From 93b97b9831b9c099c549a6f595e6df7d79050bf0 Mon Sep 17 00:00:00 2001
From: Fletcher Haynes <fletcher.haynes@molecula.com>
Date: Thu, 30 Dec 2021 17:00:59 -0800
Subject: [PATCH 19/23] Test push to see if pipeline is running on push to
 master

---
 .gitlab/.gitlab-ci.yml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/.gitlab/.gitlab-ci.yml b/.gitlab/.gitlab-ci.yml
index a27c1012e..60d1a309d 100644
--- a/.gitlab/.gitlab-ci.yml
+++ b/.gitlab/.gitlab-ci.yml
@@ -13,6 +13,7 @@ include:
 variables:
   GOVERSION: "1.16.10"
 
+
 stages:
   - lint
   - test

From 99f6a1c113f2fad7ee5e9b900e1f05d031332649 Mon Sep 17 00:00:00 2001
From: reesporte <reesedporter@gmail.com>
Date: Mon, 3 Jan 2022 10:45:23 -0600
Subject: [PATCH 20/23] change min to val

bc it could be used for things besides mins
---
 field.go | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/field.go b/field.go
index bcbc79c6e..a726514e7 100644
--- a/field.go
+++ b/field.go
@@ -1429,7 +1429,7 @@ func (f *Field) MinForShard(tx Tx, shard uint64, filter *Row) (ValCount, error)
 // includes the int64 "Val\" value to make comparisons easier in the
 // executor (at time of writing, Percentile takes advantage of this,
 // but we might be able to simplify logic in other places as well).
-func (f *Field) valCountize(min int64, cnt uint64, bsig *bsiGroup) (ValCount, error) {
+func (f *Field) valCountize(val int64, cnt uint64, bsig *bsiGroup) (ValCount, error) {
 	if bsig == nil {
 		bsig = f.bsiGroup(f.name)
 		if bsig == nil {
@@ -1440,12 +1440,12 @@ func (f *Field) valCountize(min int64, cnt uint64, bsig *bsiGroup) (ValCount, er
 	valCount := ValCount{Count: int64(cnt)}
 
 	if f.Options().Type == FieldTypeDecimal {
-		dec := pql.NewDecimal(min+bsig.Base, bsig.Scale)
+		dec := pql.NewDecimal(val+bsig.Base, bsig.Scale)
 		valCount.DecimalVal = &dec
 	} else if f.Options().Type == FieldTypeTimestamp {
-		valCount.TimestampVal = time.Unix(0, (min+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC()
+		valCount.TimestampVal = time.Unix(0, (val+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC()
 	}
-	valCount.Val = min + bsig.Base
+	valCount.Val = val + bsig.Base
 	return valCount, nil
 }
 

From fa2391b948784edd51dafa1b81151effc6113c77 Mon Sep 17 00:00:00 2001
From: reesporte <reesedporter@gmail.com>
Date: Mon, 3 Jan 2022 11:11:27 -0600
Subject: [PATCH 21/23] explicitly test untested path of valcountize

---
 field_internal_test.go | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/field_internal_test.go b/field_internal_test.go
index f7e0fadfd..d9471db91 100644
--- a/field_internal_test.go
+++ b/field_internal_test.go
@@ -182,6 +182,23 @@ func TestBSIGroup_BaseValue(t *testing.T) {
 	})
 }
 
+func TestField_ValCountize(t *testing.T) {
+	f := OpenField(t, OptFieldTypeDefault())
+	defer f.Close()
+	// check that you get an empty val count and err
+	// BSIGroupNotFound on nil bsig from
+	// f.bsiGroup(f.name)
+	f.bsiGroups = []*bsiGroup{}
+	v, err := f.valCountize(42, 42, nil)
+	if !reflect.DeepEqual(v, ValCount{}) {
+		t.Errorf("expected %v, got %v", ValCount{}, v)
+	}
+	if err != ErrBSIGroupNotFound {
+		t.Errorf("expected %v, got %v", ErrBSIGroupNotFound, err)
+	}
+
+}
+
 // Ensure field can open and retrieve a view.
 func TestField_DeleteView(t *testing.T) {
 	f := OpenField(t, OptFieldTypeDefault())

From 6e3ce01ecb6b06fb86497a2add2cf56b02dd2890 Mon Sep 17 00:00:00 2001
From: reesporte <reesedporter@gmail.com>
Date: Mon, 3 Jan 2022 11:11:54 -0600
Subject: [PATCH 22/23] explicitly test that getScaledInt works with timestamps

---
 executor_internal_test.go | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/executor_internal_test.go b/executor_internal_test.go
index 5c5ed9314..79a9cfe72 100644
--- a/executor_internal_test.go
+++ b/executor_internal_test.go
@@ -489,3 +489,18 @@ func TestExecutorSafeCopyDistinctTimestamp(t *testing.T) {
 		t.Fatalf("Did not copy results. got %+v, want %+v", copied.Results, response.Results)
 	}
 }
+
+func TestGetScaledInt(t *testing.T) {
+	f := OpenField(t, OptFieldTypeTimestamp(time.Now(), "ms"))
+	defer f.Close()
+	// check that fields with type timestamp return the int64 passed in to getScaledInt with nil err
+	v := time.Now().Unix()
+	res, err := getScaledInt(f.Field, v)
+	if err != nil {
+		t.Errorf("got error %v, expected nil", err)
+	}
+	if !reflect.DeepEqual(res, v) {
+		t.Errorf("expected %v, got %v", v, res)
+	}
+
+}

From b13538e4266aaa908ace8143526e04ee38e8ec79 Mon Sep 17 00:00:00 2001
From: reesporte <reesedporter@gmail.com>
Date: Mon, 3 Jan 2022 11:16:14 -0600
Subject: [PATCH 23/23] update doc comment

---
 field.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/field.go b/field.go
index 6435634f2..e45e82d6e 100644
--- a/field.go
+++ b/field.go
@@ -1426,7 +1426,7 @@ func (f *Field) MinForShard(tx Tx, shard uint64, filter *Row) (ValCount, error)
 	return f.valCountize(min, cnt, bsig)
 }
 
-// valCountize takes the "raw" min value and count we get from the
+// valCountize takes the "raw" value and count we get from the
 // fragment and calculates the cooked values for this field
 // (timestamping, decimaling, or just adding in the base). It always
 // includes the int64 "Val\" value to make comparisons easier in the