func newMutationPayload(t *testing.T, table string, name string, length int) func() io.Reader {
	var mutationSeq uint64
	return func() io.Reader {
		seq := atomic.AddUint64(&mutationSeq, 1)
		data := make([]byte, length)
		rand.Read(data)
		type mutation struct {
			TableName string                 `json:"table"`
			Delete    bool                   `json:"delete"`
			Values    map[string]interface{} `json:"values"`
		}
		type payload struct {
			Cookie    []byte     `json:"cookie"`
			Mutations []mutation `json:"mutations"`
		}
		cookie := make([]byte, 8)
		binary.BigEndian.PutUint64(cookie, seq)
		p := payload{
			Cookie: cookie,
			Mutations: []mutation{
				{
					TableName: table,
					Values: map[string]interface{}{
						"name": fmt.Sprintf("%s-%d", name, mutationSeq),
						"data": data,
					},
				},
			},
		}
		b, err := json.Marshal(p)
		require.NoError(t, err)
		return bytes.NewReader(b)
	}
}
newMutationPayload is a helper that produces a func that produces a reader that supplies a
payload to the mutation api. it maintains its own internal cookie that gets incremented on
each payload
package executive
import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"
	"time"
	"github.com/segmentio/ctlstore/pkg/limits"
	"github.com/segmentio/ctlstore/pkg/schema"
	"github.com/segmentio/ctlstore/pkg/units"
	"github.com/stretchr/testify/require"
)
func testDBLimiter(t *testing.T, dbType string) {
	u := newDbExecTestUtil(t, dbType)
	ctldb := u.db
	ctx, cancel := context.WithCancel(u.ctx)
	defer cancel()
	const (
		familyName     = "db_limiter_family"
		tableName      = "db_limiter_table"
		bucketInterval = 5 * time.Second
		writerLimit    = 5  
		writerName     = "db-limiter-writer-name"
		writerSecret   = "db-limiter-writer-secret"
	)
	fakeTime := newFakeTime(1000)
	defaultTableLimit := limits.SizeLimits{MaxSize: 30 * units.KILOBYTE, WarnSize: 20 * units.KILOBYTE}
	limiter := newDBLimiter(ctldb, dbType, defaultTableLimit, bucketInterval, writerLimit)
	limiter.timeFunc = fakeTime.get
	require.NoError(t, limiter.tableSizer.refresh(ctx))
	require.NoError(t, u.e.CreateFamily(familyName))
	executive := &executiveService{ctldb: u.db, ctx: ctx, limiter: limiter, serveTimeout: 10 * time.Second}
	fieldNames := []string{"name", "data"}
	fieldTypes := []schema.FieldType{schema.FTString, schema.FTBinary}
	keyFields := []string{"name"}
	require.NoError(t, u.e.CreateTable(familyName, tableName, fieldNames, fieldTypes, keyFields))
	require.NoError(t, u.e.RegisterWriter(writerName, writerSecret))
	payloadFunc := newMutationPayload(t, tableName, "test", int(10*units.KILOBYTE))
	makeMutation := func(expectedCode int) {
		req := httptest.NewRequest("POST", "/families/"+familyName+"/mutations", payloadFunc())
		req.Header.Set("ctlstore-writer", writerName)
		req.Header.Set("ctlstore-secret", writerSecret)
		w := httptest.NewRecorder()
		executive.ServeHTTP(w, req)
		resp := w.Result()
		defer resp.Body.Close()
		if expectedCode != resp.StatusCode {
			b, _ := ioutil.ReadAll(resp.Body)
			require.Failf(t, "request failed", "Expected %d, got %d: %s", expectedCode, resp.StatusCode, b)
		}
	}
	makeMutation(http.StatusOK)
	require.NoError(t, limiter.tableSizer.refresh(ctx))
	for i := 0; i < 4; i++ {
		makeMutation(http.StatusOK)
	}
	makeMutation(http.StatusTooManyRequests)
	fakeTime.add(int64(bucketInterval / time.Second))
	for i := 0; i < 5; i++ {
		makeMutation(http.StatusOK)
	}
	makeMutation(http.StatusTooManyRequests)
	_, err := u.db.ExecContext(ctx, "insert into max_writer_rates (writer_name, max_rows_per_minute) values(?,?)",
		writerName, 120)  
	require.NoError(t, err)
	require.NoError(t, limiter.refreshWriterLimits(ctx))
	for i := 0; i < 5; i++ {
		makeMutation(http.StatusOK)
	}
	makeMutation(http.StatusTooManyRequests)
	fakeTime.add(int64(bucketInterval / time.Second))
	makeMutation(http.StatusOK)
	require.NoError(t, limiter.tableSizer.refresh(ctx))
	if dbType == "sqlite3" {
		makeMutation(http.StatusOK)
	} else {
		makeMutation(http.StatusInsufficientStorage)
	}
	countRows := func() int64 {
		row := u.db.QueryRowContext(ctx, "select count(*) from writer_usage")
		var count int64
		err = row.Scan(&count)
		require.NoError(t, err)
		return count
	}
	count := countRows()
	require.True(t, count > 0, "unexpected count: %d", count)
	err = limiter.deleteOldUsageData(ctx)
	require.NoError(t, err)
	require.EqualValues(t, count, countRows())
	fakeTime.set(time.Now().Unix())
	err = limiter.deleteOldUsageData(ctx)
	require.NoError(t, err)
	count = countRows()
	require.EqualValues(t, 0, count)
}