Update vendored dependencies

This includes github.com/kurin/blazer 0.2.0, which resolves #1291

vendor/github.com/kurin/blazer/.travis.yml

@@ -7,4 +7,5 @@ branches:
   only:
   - master
  
-script: B2_LOG_LEVEL=2 go test -v ./base ./b2
+before_script: go run internal/bin/cleanup/cleanup.go
+script: go test -v ./base ./b2 ./x/...

vendor/github.com/kurin/blazer/b2/b2.go

@@ -28,6 +28,7 @@
 package b2
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"net/http"
@@ -35,8 +36,6 @@ import (
 	"strconv"
 	"sync"
 	"time"
-
-	"golang.org/x/net/context"
 )
 
 // Client is a Backblaze B2 client.
@@ -67,11 +66,23 @@ type clientOptions struct {
 	failSomeUploads bool
 	expireTokens    bool
 	capExceeded     bool
+	userAgents      []string
 }
 
 // A ClientOption allows callers to adjust various per-client settings.
 type ClientOption func(*clientOptions)
 
+// UserAgent sets the User-Agent HTTP header.  The default header is
+// "blazer/<version>"; the value set here will be prepended to that.  This can
+// be set multiple times.
+//
+// A user agent is generally of the form "<product>/<version> (<comments>)".
+func UserAgent(agent string) ClientOption {
+	return func(o *clientOptions) {
+		o.userAgents = append(o.userAgents, agent)
+	}
+}
+
 // Transport sets the underlying HTTP transport mechanism.  If unset,
 // http.DefaultTransport is used.
 func Transport(rt http.RoundTripper) ClientOption {
@@ -118,6 +129,7 @@ const (
 	UnknownType BucketType = ""
 	Private                = "allPrivate"
 	Public                 = "allPublic"
+	Snapshot               = "snapshot"
 )
 
 // BucketAttrs holds a bucket's metadata attributes.
@@ -582,11 +594,19 @@ func (b *Bucket) Reveal(ctx context.Context, name string) error {
 	return obj.Delete(ctx)
 }
 
+// I don't want to import all of ioutil for this.
+type discard struct{}
+
+func (discard) Write(p []byte) (int, error) {
+	return len(p), nil
+}
+
 func (b *Bucket) getObject(ctx context.Context, name string) (*Object, error) {
 	fr, err := b.b.downloadFileByName(ctx, name, 0, 1)
 	if err != nil {
 		return nil, err
 	}
+	io.Copy(discard{}, fr)
 	fr.Close()
 	return &Object{
 		name: name,

vendor/github.com/kurin/blazer/b2/b2_test.go

@@ -16,6 +16,7 @@ package b2
 
 import (
 	"bytes"
+	"context"
 	"crypto/sha1"
 	"fmt"
 	"io"
@@ -26,8 +27,6 @@ import (
 	"sync"
 	"testing"
 	"time"
-
-	"golang.org/x/net/context"
 )
 
 const (
@@ -352,6 +351,82 @@ func (zReader) Read(p []byte) (int, error) {
 	return len(p), nil
 }
 
+type zReadSeeker struct {
+	size int64
+	pos  int64
+}
+
+func (rs *zReadSeeker) Read(p []byte) (int, error) {
+	for i := rs.pos; ; i++ {
+		j := int(i - rs.pos)
+		if j >= len(p) || i >= rs.size {
+			var rtn error
+			if i >= rs.size {
+				rtn = io.EOF
+			}
+			rs.pos = i
+			return j, rtn
+		}
+		f := int(i) % len(pattern)
+		p[j] = pattern[f]
+	}
+}
+
+func (rs *zReadSeeker) Seek(offset int64, whence int) (int64, error) {
+	switch whence {
+	case io.SeekStart:
+		rs.pos = offset
+	case io.SeekEnd:
+		rs.pos = rs.size + offset
+	}
+	return rs.pos, nil
+}
+
+func TestReaderFrom(t *testing.T) {
+	ctx := context.Background()
+	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+
+	table := []struct {
+		size, pos int64
+	}{
+		{
+			size: 10,
+		},
+	}
+
+	for _, e := range table {
+		client := &Client{
+			backend: &beRoot{
+				b2i: &testRoot{
+					bucketMap: make(map[string]map[string]string),
+					errs:      &errCont{},
+				},
+			},
+		}
+
+		bucket, err := client.NewBucket(ctx, bucketName, &BucketAttrs{Type: Private})
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() {
+			if err := bucket.Delete(ctx); err != nil {
+				t.Error(err)
+			}
+		}()
+
+		r := &zReadSeeker{pos: e.pos, size: e.size}
+		w := bucket.Object("writer").NewWriter(ctx)
+		n, err := w.ReadFrom(r)
+		if err != nil {
+			t.Errorf("ReadFrom(): %v", err)
+		}
+		if n != e.size {
+			t.Errorf("ReadFrom(): got %d bytes, wanted %d bytes", n, e.size)
+		}
+	}
+}
+
 func TestReauth(t *testing.T) {
 	ctx := context.Background()
 	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
@@ -696,6 +771,52 @@ func TestFileBuffer(t *testing.T) {
 	}
 }
 
+func TestNonBuffer(t *testing.T) {
+	table := []struct {
+		str  string
+		off  int64
+		len  int64
+		want string
+	}{
+		{
+			str:  "a string",
+			off:  0,
+			len:  3,
+			want: "a s",
+		},
+		{
+			str:  "a string",
+			off:  3,
+			len:  1,
+			want: "t",
+		},
+		{
+			str:  "a string",
+			off:  3,
+			len:  5,
+			want: "tring",
+		},
+	}
+
+	for _, e := range table {
+		nb := newNonBuffer(strings.NewReader(e.str), e.off, e.len)
+		want := fmt.Sprintf("%s%x", e.want, sha1.Sum([]byte(e.str[int(e.off):int(e.off+e.len)])))
+		r, err := nb.Reader()
+		if err != nil {
+			t.Error(err)
+			continue
+		}
+		got, err := ioutil.ReadAll(r)
+		if err != nil {
+			t.Errorf("ioutil.ReadAll(%#v): %v", e, err)
+			continue
+		}
+		if want != string(got) {
+			t.Errorf("ioutil.ReadAll(%#v): got %q, want %q", e, string(got), want)
+		}
+	}
+}
+
 func writeFile(ctx context.Context, bucket *Bucket, name string, size int64, csize int) (*Object, string, error) {
 	r := io.LimitReader(zReader{}, size)
 	o := bucket.Object(name)
@@ -704,9 +825,13 @@ func writeFile(ctx context.Context, bucket *Bucket, name string, size int64, csi
 	w := io.MultiWriter(f, h)
 	f.ConcurrentUploads = 5
 	f.ChunkSize = csize
-	if _, err := io.Copy(w, r); err != nil {
+	n, err := io.Copy(w, r)
+	if err != nil {
 		return nil, "", err
 	}
+	if n != size {
+		return nil, "", fmt.Errorf("io.Copy(): wrote %d bytes; wanted %d bytes", n, size)
+	}
 	if err := f.Close(); err != nil {
 		return nil, "", err
 	}

vendor/github.com/kurin/blazer/b2/backend.go

@@ -15,11 +15,10 @@
 package b2
 
 import (
+	"context"
 	"io"
 	"math/rand"
 	"time"
-
-	"golang.org/x/net/context"
 )
 
 // This file wraps the baseline interfaces with backoff and retry semantics.
@@ -63,7 +62,7 @@ type beBucket struct {
 }
 
 type beURLInterface interface {
-	uploadFile(context.Context, io.ReadSeeker, int, string, string, string, map[string]string) (beFileInterface, error)
+	uploadFile(context.Context, readResetter, int, string, string, string, map[string]string) (beFileInterface, error)
 }
 
 type beURL struct {
@@ -100,7 +99,7 @@ type beLargeFile struct {
 
 type beFileChunkInterface interface {
 	reload(context.Context) error
-	uploadPart(context.Context, io.ReadSeeker, string, int, int) (int, error)
+	uploadPart(context.Context, readResetter, string, int, int) (int, error)
 }
 
 type beFileChunk struct {
@@ -414,10 +413,10 @@ func (b *beBucket) file(id, name string) beFileInterface {
 	}
 }
 
-func (b *beURL) uploadFile(ctx context.Context, r io.ReadSeeker, size int, name, ct, sha1 string, info map[string]string) (beFileInterface, error) {
+func (b *beURL) uploadFile(ctx context.Context, r readResetter, size int, name, ct, sha1 string, info map[string]string) (beFileInterface, error) {
 	var file beFileInterface
 	f := func() error {
-		if _, err := r.Seek(0, 0); err != nil {
+		if err := r.Reset(); err != nil {
 			return err
 		}
 		f, err := b.b2url.uploadFile(ctx, r, size, name, ct, sha1, info)
@@ -578,12 +577,12 @@ func (b *beFileChunk) reload(ctx context.Context) error {
 	return withBackoff(ctx, b.ri, f)
 }
 
-func (b *beFileChunk) uploadPart(ctx context.Context, r io.ReadSeeker, sha1 string, size, index int) (int, error) {
+func (b *beFileChunk) uploadPart(ctx context.Context, r readResetter, sha1 string, size, index int) (int, error) {
 	// no re-auth; pass it back up to the caller so they can get an new upload URI and token
 	// TODO: we should handle that here probably
 	var i int
 	f := func() error {
-		if _, err := r.Seek(0, 0); err != nil {
+		if err := r.Reset(); err != nil {
 			return err
 		}
 		j, err := b.b2fileChunk.uploadPart(ctx, r, sha1, size, index)

vendor/github.com/kurin/blazer/b2/baseline.go

@@ -15,13 +15,12 @@
 package b2
 
 import (
+	"context"
 	"io"
 	"net/http"
 	"time"
 
 	"github.com/kurin/blazer/base"
-
-	"golang.org/x/net/context"
 )
 
 // This file wraps the base package in a thin layer, for testing.  It should be
@@ -150,6 +149,9 @@ func (b *b2Root) authorizeAccount(ctx context.Context, account, key string, opts
 	if c.capExceeded {
 		aopts = append(aopts, base.ForceCapExceeded())
 	}
+	for _, agent := range c.userAgents {
+		aopts = append(aopts, base.UserAgent(agent))
+	}
 	nb, err := base.AuthorizeAccount(ctx, account, key, aopts...)
 	if err != nil {
 		return err

vendor/github.com/kurin/blazer/b2/buffer.go

@@ -17,22 +17,84 @@ package b2
 import (
 	"bytes"
 	"crypto/sha1"
+	"errors"
 	"fmt"
 	"hash"
 	"io"
 	"io/ioutil"
 	"os"
+	"strings"
 	"sync"
 )
 
+type readResetter interface {
+	Read([]byte) (int, error)
+	Reset() error
+}
+
+type resetter struct {
+	rs io.ReadSeeker
+}
+
+func (r resetter) Read(p []byte) (int, error) { return r.rs.Read(p) }
+func (r resetter) Reset() error               { _, err := r.rs.Seek(0, 0); return err }
+
+func newResetter(p []byte) readResetter { return resetter{rs: bytes.NewReader(p)} }
+
 type writeBuffer interface {
 	io.Writer
 	Len() int
-	Reader() (io.ReadSeeker, error)
+	Reader() (readResetter, error)
 	Hash() string // sha1 or whatever it is
 	Close() error
 }
 
+// nonBuffer doesn't buffer anything, but passes values directly from the
+// source readseeker.  Many nonBuffers can point at different parts of the same
+// underlying source, and be accessed by multiple goroutines simultaneously.
+func newNonBuffer(rs io.ReaderAt, offset, size int64) writeBuffer {
+	return &nonBuffer{
+		r:    io.NewSectionReader(rs, offset, size),
+		size: int(size),
+		hsh:  sha1.New(),
+	}
+}
+
+type nonBuffer struct {
+	r    *io.SectionReader
+	size int
+	hsh  hash.Hash
+
+	isEOF bool
+	buf   *strings.Reader
+}
+
+func (nb *nonBuffer) Len() int                      { return nb.size + 40 }
+func (nb *nonBuffer) Hash() string                  { return "hex_digits_at_end" }
+func (nb *nonBuffer) Close() error                  { return nil }
+func (nb *nonBuffer) Reader() (readResetter, error) { return nb, nil }
+func (nb *nonBuffer) Write([]byte) (int, error)     { return 0, errors.New("writes not supported") }
+
+func (nb *nonBuffer) Read(p []byte) (int, error) {
+	if nb.isEOF {
+		return nb.buf.Read(p)
+	}
+	n, err := io.TeeReader(nb.r, nb.hsh).Read(p)
+	if err == io.EOF {
+		err = nil
+		nb.isEOF = true
+		nb.buf = strings.NewReader(fmt.Sprintf("%x", nb.hsh.Sum(nil)))
+	}
+	return n, err
+}
+
+func (nb *nonBuffer) Reset() error {
+	nb.hsh.Reset()
+	nb.isEOF = false
+	_, err := nb.r.Seek(0, 0)
+	return err
+}
+
 type memoryBuffer struct {
 	buf *bytes.Buffer
 	hsh hash.Hash
@@ -56,15 +118,10 @@ func newMemoryBuffer() *memoryBuffer {
 	return mb
 }
 
-type thing struct {
-	rs io.ReadSeeker
-	t  int
-}
-
-func (mb *memoryBuffer) Write(p []byte) (int, error)    { return mb.w.Write(p) }
-func (mb *memoryBuffer) Len() int                       { return mb.buf.Len() }
-func (mb *memoryBuffer) Reader() (io.ReadSeeker, error) { return bytes.NewReader(mb.buf.Bytes()), nil }
-func (mb *memoryBuffer) Hash() string                   { return fmt.Sprintf("%x", mb.hsh.Sum(nil)) }
+func (mb *memoryBuffer) Write(p []byte) (int, error)   { return mb.w.Write(p) }
+func (mb *memoryBuffer) Len() int                      { return mb.buf.Len() }
+func (mb *memoryBuffer) Reader() (readResetter, error) { return newResetter(mb.buf.Bytes()), nil }
+func (mb *memoryBuffer) Hash() string                  { return fmt.Sprintf("%x", mb.hsh.Sum(nil)) }
 
 func (mb *memoryBuffer) Close() error {
 	mb.mux.Lock()
@@ -107,7 +164,7 @@ func (fb *fileBuffer) Write(p []byte) (int, error) {
 func (fb *fileBuffer) Len() int     { return fb.s }
 func (fb *fileBuffer) Hash() string { return fmt.Sprintf("%x", fb.hsh.Sum(nil)) }
 
-func (fb *fileBuffer) Reader() (io.ReadSeeker, error) {
+func (fb *fileBuffer) Reader() (readResetter, error) {
 	if _, err := fb.f.Seek(0, 0); err != nil {
 		return nil, err
 	}
@@ -124,5 +181,5 @@ type fr struct {
 	f *os.File
 }
 
-func (r *fr) Read(p []byte) (int, error)         { return r.f.Read(p) }
-func (r *fr) Seek(a int64, b int) (int64, error) { return r.f.Seek(a, b) }
+func (r *fr) Read(p []byte) (int, error) { return r.f.Read(p) }
+func (r *fr) Reset() error               { _, err := r.f.Seek(0, 0); return err }

vendor/github.com/kurin/blazer/b2/integration_test.go

@@ -16,6 +16,7 @@ package b2
 
 import (
 	"bytes"
+	"context"
 	"crypto/sha1"
 	"fmt"
 	"io"
@@ -23,10 +24,9 @@ import (
 	"os"
 	"reflect"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
-
-	"golang.org/x/net/context"
 )
 
 const (
@@ -79,6 +79,86 @@ func TestReadWriteLive(t *testing.T) {
 	}
 }
 
+func TestReaderFromLive(t *testing.T) {
+	ctx := context.Background()
+	ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
+	defer cancel()
+
+	bucket, done := startLiveTest(ctx, t)
+	defer done()
+
+	table := []struct {
+		size, pos      int64
+		csize, writers int
+	}{
+		{
+			// that it works at all
+			size: 10,
+		},
+		{
+			// large uploads
+			size:    15e6 + 10,
+			csize:   5e6,
+			writers: 2,
+		},
+		{
+			// an excess of writers
+			size:    50e6,
+			csize:   5e6,
+			writers: 12,
+		},
+		{
+			// with offset, seeks back to start after turning it into a ReaderAt
+			size: 250,
+			pos:  50,
+		},
+	}
+
+	for i, e := range table {
+		rs := &zReadSeeker{pos: e.pos, size: e.size}
+		o := bucket.Object(fmt.Sprintf("writer.%d", i))
+		w := o.NewWriter(ctx)
+		w.ChunkSize = e.csize
+		w.ConcurrentUploads = e.writers
+		n, err := w.ReadFrom(rs)
+		if err != nil {
+			t.Errorf("ReadFrom(): %v", err)
+		}
+		if n != e.size {
+			t.Errorf("ReadFrom(): got %d bytes, wanted %d bytes", n, e.size)
+		}
+		if err := w.Close(); err != nil {
+			t.Errorf("w.Close(): %v", err)
+			continue
+		}
+
+		r := o.NewReader(ctx)
+		h := sha1.New()
+		rn, err := io.Copy(h, r)
+		if err != nil {
+			t.Errorf("Read from B2: %v", err)
+		}
+		if rn != n {
+			t.Errorf("Read from B2: got %d bytes, want %d bytes", rn, n)
+		}
+		if err := r.Close(); err != nil {
+			t.Errorf("r.Close(): %v", err)
+		}
+
+		hex := fmt.Sprintf("%x", h.Sum(nil))
+		attrs, err := o.Attrs(ctx)
+		if err != nil {
+			t.Errorf("Attrs(): %v", err)
+			continue
+		}
+		if attrs.SHA1 == "none" {
+			continue
+		}
+		if hex != attrs.SHA1 {
+			t.Errorf("SHA1: got %q, want %q", hex, attrs.SHA1)
+		}
+	}
+}
 func TestHideShowLive(t *testing.T) {
 	ctx := context.Background()
 	ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
@@ -200,7 +280,6 @@ func TestResumeWriter(t *testing.T) {
 }
 
 func TestAttrs(t *testing.T) {
-	// TODO: test is flaky
 	ctx := context.Background()
 	ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
 	defer cancel()
@@ -765,6 +844,74 @@ func listObjects(ctx context.Context, f func(context.Context, int, *Cursor) ([]*
 
 var transport = http.DefaultTransport
 
+type eofTripper struct {
+	rt http.RoundTripper
+	t  *testing.T
+}
+
+func (et eofTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	resp, err := et.rt.RoundTrip(req)
+	if err != nil {
+		return nil, err
+	}
+	resp.Body = &eofReadCloser{rc: resp.Body, t: et.t}
+	return resp, nil
+}
+
+type eofReadCloser struct {
+	rc  io.ReadCloser
+	eof bool
+	t   *testing.T
+}
+
+func (eof *eofReadCloser) Read(p []byte) (int, error) {
+	n, err := eof.rc.Read(p)
+	if err == io.EOF {
+		eof.eof = true
+	}
+	return n, err
+}
+
+func (eof *eofReadCloser) Close() error {
+	if !eof.eof {
+		eof.t.Error("http body closed with bytes unread")
+	}
+	return eof.rc.Close()
+}
+
+// Checks that close is called.
+type ccTripper struct {
+	t     *testing.T
+	rt    http.RoundTripper
+	trips int64
+}
+
+func (cc *ccTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	resp, err := cc.rt.RoundTrip(req)
+	if err != nil {
+		return nil, err
+	}
+	atomic.AddInt64(&cc.trips, 1)
+	resp.Body = &ccRC{ReadCloser: resp.Body, c: &cc.trips}
+	return resp, err
+}
+
+func (cc *ccTripper) done() {
+	if cc.trips != 0 {
+		cc.t.Errorf("failed to close %d HTTP bodies", cc.trips)
+	}
+}
+
+type ccRC struct {
+	io.ReadCloser
+	c *int64
+}
+
+func (cc *ccRC) Close() error {
+	atomic.AddInt64(cc.c, -1)
+	return cc.ReadCloser.Close()
+}
+
 func startLiveTest(ctx context.Context, t *testing.T) (*Bucket, func()) {
 	id := os.Getenv(apiID)
 	key := os.Getenv(apiKey)
@@ -772,7 +919,9 @@ func startLiveTest(ctx context.Context, t *testing.T) (*Bucket, func()) {
 		t.Skipf("B2_ACCOUNT_ID or B2_SECRET_KEY unset; skipping integration tests")
 		return nil, nil
 	}
-	client, err := NewClient(ctx, id, key, FailSomeUploads(), ExpireSomeAuthTokens(), Transport(transport))
+	ccport := &ccTripper{rt: transport, t: t}
+	tport := eofTripper{rt: ccport, t: t}
+	client, err := NewClient(ctx, id, key, FailSomeUploads(), ExpireSomeAuthTokens(), Transport(tport), UserAgent("b2-test"), UserAgent("integration-test"))
 	if err != nil {
 		t.Fatal(err)
 		return nil, nil
@@ -783,6 +932,7 @@ func startLiveTest(ctx context.Context, t *testing.T) (*Bucket, func()) {
 		return nil, nil
 	}
 	f := func() {
+		defer ccport.done()
 		for c := range listObjects(ctx, bucket.ListObjects) {
 			if c.err != nil {
 				continue

vendor/github.com/kurin/blazer/b2/reader.go

@@ -16,13 +16,12 @@ package b2
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"io"
 	"sync"
 
 	"github.com/kurin/blazer/internal/blog"
-
-	"golang.org/x/net/context"
 )
 
 var errNoMoreContent = errors.New("416: out of content")
@@ -140,11 +139,12 @@ func (r *Reader) thread() {
 				return
 			}
 			rsize, _, _, _ := fr.stats()
-			mr := &meteredReader{r: &fakeSeeker{fr}, size: int(rsize)}
+			mr := &meteredReader{r: noopResetter{fr}, size: int(rsize)}
 			r.smux.Lock()
 			r.smap[chunkID] = mr
 			r.smux.Unlock()
 			i, err := copyContext(r.ctx, buf, mr)
+			fr.Close()
 			r.smux.Lock()
 			r.smap[chunkID] = nil
 			r.smux.Unlock()
@@ -290,11 +290,8 @@ func copyContext(ctx context.Context, dst io.Writer, src io.Reader) (written int
 	return written, err
 }
 
-// fakeSeeker exists so that we can wrap the http response body (an io.Reader
-// but not an io.Seeker) into a meteredReader, which will allow us to keep tabs
-// on how much of the chunk we've read so far.
-type fakeSeeker struct {
+type noopResetter struct {
 	io.Reader
 }
 
-func (fs *fakeSeeker) Seek(int64, int) (int64, error) { return 0, nil }
+func (noopResetter) Reset() error { return nil }

vendor/github.com/kurin/blazer/b2/readerat.go

@@ -0,0 +1,48 @@
+// Copyright 2017, Google
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package b2
+
+import (
+	"io"
+	"sync"
+)
+
+type readerAt struct {
+	rs io.ReadSeeker
+	mu sync.Mutex
+}
+
+func (r *readerAt) ReadAt(p []byte, off int64) (int, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	// ReadAt is supposed to preserve the offset.
+	cur, err := r.rs.Seek(0, io.SeekCurrent)
+	if err != nil {
+		return 0, err
+	}
+	defer func() { r.rs.Seek(cur, io.SeekStart) }()
+
+	if _, err := r.rs.Seek(off, io.SeekStart); err != nil {
+		return 0, err
+	}
+	return io.ReadFull(r.rs, p)
+}
+
+// wraps a ReadSeeker in a mutex to provite a ReaderAt how is this not in the
+// io package?
+func enReaderAt(rs io.ReadSeeker) io.ReaderAt {
+	return &readerAt{rs: rs}
+}

vendor/github.com/kurin/blazer/b2/writer.go

@@ -15,6 +15,7 @@
 package b2
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -23,8 +24,6 @@ import (
 	"time"
 
 	"github.com/kurin/blazer/internal/blog"
-
-	"golang.org/x/net/context"
 )
 
 // Writer writes data into Backblaze.  It automatically switches to the large
@@ -75,6 +74,7 @@ type Writer struct {
 	file        beLargeFileInterface
 	seen        map[int]string
 	everStarted bool
+	newBuffer   func() (writeBuffer, error)
 
 	o    *Object
 	name string
@@ -94,15 +94,8 @@ type chunk struct {
 	buf writeBuffer
 }
 
-func (w *Writer) getBuffer() (writeBuffer, error) {
-	if !w.UseFileBuffer {
-		return newMemoryBuffer(), nil
-	}
-	return newFileBuffer(w.FileBufferDir)
-}
-
 func (w *Writer) setErr(err error) {
-	if err == nil {
+	if err == nil || err == io.EOF {
 		return
 	}
 	w.emux.Lock()
@@ -200,8 +193,7 @@ func (w *Writer) thread() {
 	}()
 }
 
-// Write satisfies the io.Writer interface.
-func (w *Writer) Write(p []byte) (int, error) {
+func (w *Writer) init() {
 	w.start.Do(func() {
 		w.everStarted = true
 		w.smux.Lock()
@@ -212,13 +204,24 @@ func (w *Writer) Write(p []byte) (int, error) {
 		if w.csize == 0 {
 			w.csize = 1e8
 		}
-		v, err := w.getBuffer()
+		if w.newBuffer == nil {
+			w.newBuffer = func() (writeBuffer, error) { return newMemoryBuffer(), nil }
+			if w.UseFileBuffer {
+				w.newBuffer = func() (writeBuffer, error) { return newFileBuffer(w.FileBufferDir) }
+			}
+		}
+		v, err := w.newBuffer()
 		if err != nil {
 			w.setErr(err)
 			return
 		}
 		w.w = v
 	})
+}
+
+// Write satisfies the io.Writer interface.
+func (w *Writer) Write(p []byte) (int, error) {
+	w.init()
 	if err := w.getErr(); err != nil {
 		return 0, err
 	}
@@ -263,7 +266,7 @@ redo:
 	f, err := ue.uploadFile(w.ctx, mr, int(w.w.Len()), w.name, ctype, sha1, w.info)
 	if err != nil {
 		if w.o.b.r.reupload(err) {
-			blog.V(1).Infof("b2 writer: %v; retrying", err)
+			blog.V(2).Infof("b2 writer: %v; retrying", err)
 			u, err := w.o.b.b.getUploadURL(w.ctx)
 			if err != nil {
 				return err
@@ -352,7 +355,7 @@ func (w *Writer) sendChunk() error {
 		return w.ctx.Err()
 	}
 	w.cidx++
-	v, err := w.getBuffer()
+	v, err := w.newBuffer()
 	if err != nil {
 		return err
 	}
@@ -360,15 +363,83 @@ func (w *Writer) sendChunk() error {
 	return nil
 }
 
+// ReadFrom reads all of r into w, returning the first error or no error if r
+// returns io.EOF.  If r is also an io.Seeker, ReadFrom will stream r directly
+// over the wire instead of buffering it locally.  This reduces memory usage.
+//
+// Do not issue multiple calls to ReadFrom, or mix ReadFrom and Write.  If you
+// have multiple readers you want to concatenate into the same B2 object, use
+// an io.MultiReader.
+//
+// Note that io.Copy will automatically choose to use ReadFrom.
+//
+// ReadFrom currently doesn't handle w.Resume; if w.Resume is true, ReadFrom
+// will act as if r is not an io.Seeker.
+func (w *Writer) ReadFrom(r io.Reader) (int64, error) {
+	rs, ok := r.(io.ReadSeeker)
+	if !ok || w.Resume {
+		return copyContext(w.ctx, w, r)
+	}
+	blog.V(2).Info("streaming without buffer")
+	size, err := rs.Seek(0, io.SeekEnd)
+	if err != nil {
+		return 0, err
+	}
+	var ra io.ReaderAt
+	if rat, ok := r.(io.ReaderAt); ok {
+		ra = rat
+	} else {
+		ra = enReaderAt(rs)
+	}
+	var offset int64
+	var wrote int64
+	w.newBuffer = func() (writeBuffer, error) {
+		left := size - offset
+		if left <= 0 {
+			// We're done sending real chunks; send empty chunks from now on so that
+			// Close() works.
+			w.newBuffer = func() (writeBuffer, error) { return newMemoryBuffer(), nil }
+			w.w = newMemoryBuffer()
+			return nil, io.EOF
+		}
+		csize := int64(w.csize)
+		if left < csize {
+			csize = left
+		}
+		nb := newNonBuffer(ra, offset, csize)
+		wrote += csize // TODO: this is kind of a total lie
+		offset += csize
+		return nb, nil
+	}
+	w.init()
+	if size < int64(w.csize) {
+		// the magic happens on w.Close()
+		return size, nil
+	}
+	for {
+		if err := w.sendChunk(); err != nil {
+			if err != io.EOF {
+				return wrote, err
+			}
+			return wrote, nil
+		}
+	}
+}
+
 // Close satisfies the io.Closer interface.  It is critical to check the return
-// value of Close on all writers.
+// value of Close for all writers.
 func (w *Writer) Close() error {
 	w.done.Do(func() {
 		if !w.everStarted {
 			return
 		}
 		defer w.o.b.c.removeWriter(w)
-		defer w.w.Close() // TODO: log error
+		defer func() {
+			if err := w.w.Close(); err != nil {
+				// this is non-fatal, but alarming
+				blog.V(1).Infof("close %s: %v", w.name, err)
+			}
+		}()
 		if w.cidx == 0 {
 			w.setErr(w.simpleWriteFile())
 			return
@@ -423,7 +494,7 @@ func (w *Writer) status() *WriterStatus {
 type meteredReader struct {
 	read int64
 	size int
-	r    io.ReadSeeker
+	r    readResetter
 	mux  sync.Mutex
 }
 
@@ -435,11 +506,11 @@ func (mr *meteredReader) Read(p []byte) (int, error) {
 	return n, err
 }
 
-func (mr *meteredReader) Seek(offset int64, whence int) (int64, error) {
+func (mr *meteredReader) Reset() error {
 	mr.mux.Lock()
 	defer mr.mux.Unlock()
-	mr.read = offset
-	return mr.r.Seek(offset, whence)
+	mr.read = 0
+	return mr.r.Reset()
 }
 
 func (mr *meteredReader) done() float64 {

vendor/github.com/kurin/blazer/base/base.go

@@ -23,6 +23,7 @@ package base
 
 import (
 	"bytes"
+	"context"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
@@ -38,12 +39,11 @@ import (
 
 	"github.com/kurin/blazer/internal/b2types"
 	"github.com/kurin/blazer/internal/blog"
-
-	"golang.org/x/net/context"
 )
 
-var (
-	APIBase = "https://api.backblazeb2.com"
+const (
+	APIBase          = "https://api.backblazeb2.com"
+	DefaultUserAgent = "blazer/0.1.1"
 )
 
 type b2err struct {
@@ -218,6 +218,22 @@ type b2Options struct {
 	failSomeUploads bool
 	expireTokens    bool
 	capExceeded     bool
+	apiBase         string
+	userAgent       string
+}
+
+func (o *b2Options) getAPIBase() string {
+	if o.apiBase != "" {
+		return o.apiBase
+	}
+	return APIBase
+}
+
+func (o *b2Options) getUserAgent() string {
+	if o.userAgent != "" {
+		return fmt.Sprintf("%s %s", o.userAgent, DefaultUserAgent)
+	}
+	return DefaultUserAgent
 }
 
 func (o *b2Options) getTransport() http.RoundTripper {
@@ -281,6 +297,34 @@ func (rb *requestBody) getBody() io.Reader {
 	return rb.body
 }
 
+type keepFinalBytes struct {
+	r      io.Reader
+	remain int
+	sha    [40]byte
+}
+
+func (k *keepFinalBytes) Read(p []byte) (int, error) {
+	n, err := k.r.Read(p)
+	if k.remain-n > 40 {
+		k.remain -= n
+		return n, err
+	}
+	// This was a whole lot harder than it looks.
+	pi := -40 + k.remain
+	if pi < 0 {
+		pi = 0
+	}
+	pe := n
+	ki := 40 - k.remain
+	if ki < 0 {
+		ki = 0
+	}
+	ke := n - k.remain + 40
+	copy(k.sha[ki:ke], p[pi:pe])
+	k.remain -= n
+	return n, err
+}
+
 var reqID int64
 
 func (o *b2Options) makeRequest(ctx context.Context, method, verb, uri string, b2req, b2resp interface{}, headers map[string]string, body *requestBody) error {
@@ -307,6 +351,7 @@ func (o *b2Options) makeRequest(ctx context.Context, method, verb, uri string, b
 		}
 		req.Header.Set(k, v)
 	}
+	req.Header.Set("User-Agent", o.getUserAgent())
 	req.Header.Set("X-Blazer-Request-ID", fmt.Sprintf("%d", atomic.AddInt64(&reqID, 1)))
 	req.Header.Set("X-Blazer-Method", method)
 	if o.failSomeUploads {
@@ -372,7 +417,7 @@ func AuthorizeAccount(ctx context.Context, account, key string, opts ...AuthOpti
 	for _, f := range opts {
 		f(b2opts)
 	}
-	if err := b2opts.makeRequest(ctx, "b2_authorize_account", "GET", APIBase+b2types.V1api+"b2_authorize_account", nil, b2resp, headers, nil); err != nil {
+	if err := b2opts.makeRequest(ctx, "b2_authorize_account", "GET", b2opts.getAPIBase()+b2types.V1api+"b2_authorize_account", nil, b2resp, headers, nil); err != nil {
 		return nil, err
 	}
 	return &B2{
@@ -388,6 +433,19 @@ func AuthorizeAccount(ctx context.Context, account, key string, opts ...AuthOpti
 // An AuthOption allows callers to choose per-session settings.
 type AuthOption func(*b2Options)
 
+// UserAgent sets the User-Agent HTTP header.  The default header is
+// "blazer/<version>"; the value set here will be prepended to that.  This can
+// be set multiple times.
+func UserAgent(agent string) AuthOption {
+	return func(o *b2Options) {
+		if o.userAgent == "" {
+			o.userAgent = agent
+			return
+		}
+		o.userAgent = fmt.Sprintf("%s %s", agent, o.userAgent)
+	}
+}
+
 // Transport returns an AuthOption that sets the underlying HTTP mechanism.
 func Transport(rt http.RoundTripper) AuthOption {
 	return func(o *b2Options) {
@@ -817,10 +875,16 @@ func (fc *FileChunk) UploadPart(ctx context.Context, r io.Reader, sha1 string, s
 		"Content-Length":    fmt.Sprintf("%d", size),
 		"X-Bz-Content-Sha1": sha1,
 	}
+	if sha1 == "hex_digits_at_end" {
+		r = &keepFinalBytes{r: r, remain: size}
+	}
 	if err := fc.file.b2.opts.makeRequest(ctx, "b2_upload_part", "POST", fc.url, nil, nil, headers, &requestBody{body: r, size: int64(size)}); err != nil {
 		return 0, err
 	}
 	fc.file.mu.Lock()
+	if sha1 == "hex_digits_at_end" {
+		sha1 = string(r.(*keepFinalBytes).sha[:])
+	}
 	fc.file.hashes[index] = sha1
 	fc.file.size += int64(size)
 	fc.file.mu.Unlock()
@@ -1003,35 +1067,36 @@ func (b *Bucket) DownloadFileByName(ctx context.Context, name string, offset, si
 	resp := reply.resp
 	logResponse(resp, nil)
 	if resp.StatusCode != 200 && resp.StatusCode != 206 {
+		defer resp.Body.Close()
 		return nil, mkErr(resp)
 	}
-	clen, err := strconv.ParseInt(reply.resp.Header.Get("Content-Length"), 10, 64)
+	clen, err := strconv.ParseInt(resp.Header.Get("Content-Length"), 10, 64)
 	if err != nil {
-		reply.resp.Body.Close()
+		resp.Body.Close()
 		return nil, err
 	}
 	info := make(map[string]string)
-	for key := range reply.resp.Header {
+	for key := range resp.Header {
 		if !strings.HasPrefix(key, "X-Bz-Info-") {
 			continue
 		}
 		name, err := unescape(strings.TrimPrefix(key, "X-Bz-Info-"))
 		if err != nil {
-			reply.resp.Body.Close()
+			resp.Body.Close()
 			return nil, err
 		}
-		val, err := unescape(reply.resp.Header.Get(key))
+		val, err := unescape(resp.Header.Get(key))
 		if err != nil {
-			reply.resp.Body.Close()
+			resp.Body.Close()
 			return nil, err
 		}
 		info[name] = val
 	}
 	return &FileReader{
-		ReadCloser:    reply.resp.Body,
-		SHA1:          reply.resp.Header.Get("X-Bz-Content-Sha1"),
-		ID:            reply.resp.Header.Get("X-Bz-File-Id"),
-		ContentType:   reply.resp.Header.Get("Content-Type"),
+		ReadCloser:    resp.Body,
+		SHA1:          resp.Header.Get("X-Bz-Content-Sha1"),
+		ID:            resp.Header.Get("X-Bz-File-Id"),
+		ContentType:   resp.Header.Get("Content-Type"),
 		ContentLength: int(clen),
 		Info:          info,
 	}, nil

vendor/github.com/kurin/blazer/base/integration_test.go

@@ -57,7 +57,7 @@ func TestStorage(t *testing.T) {
 	ctx := context.Background()
 
 	// b2_authorize_account
-	b2, err := AuthorizeAccount(ctx, id, key)
+	b2, err := AuthorizeAccount(ctx, id, key, UserAgent("blazer-base-test"))
 	if err != nil {
 		t.Fatal(err)
 	}

vendor/github.com/kurin/blazer/internal/bin/cleanup/cleanup.go

@@ -0,0 +1,63 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+
+	"github.com/kurin/blazer/b2"
+)
+
+const (
+	apiID  = "B2_ACCOUNT_ID"
+	apiKey = "B2_SECRET_KEY"
+)
+
+func main() {
+	id := os.Getenv(apiID)
+	key := os.Getenv(apiKey)
+	ctx := context.Background()
+	client, err := b2.NewClient(ctx, id, key)
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	var wg sync.WaitGroup
+	for _, name := range []string{"consistobucket", "base-tests"} {
+		wg.Add(1)
+		go func(name string) {
+			defer wg.Done()
+			if err := killBucket(ctx, client, id, name); err != nil {
+				fmt.Println(err)
+			}
+		}(name)
+	}
+	wg.Wait()
+}
+
+func killBucket(ctx context.Context, client *b2.Client, id, name string) error {
+	bucket, err := client.NewBucket(ctx, id+"-"+name, nil)
+	if b2.IsNotExist(err) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	defer bucket.Delete(ctx)
+	cur := &b2.Cursor{}
+	for {
+		os, c, err := bucket.ListObjects(ctx, 1000, cur)
+		if err != nil && err != io.EOF {
+			return err
+		}
+		for _, o := range os {
+			o.Delete(ctx)
+		}
+		if err == io.EOF {
+			return nil
+		}
+		cur = c
+	}
+}

vendor/github.com/kurin/blazer/x/consistent/consistent.go

@@ -0,0 +1,363 @@
+// Copyright 2016, Google
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package consistent implements an experimental interface for using B2 as a
+// coordination primitive.
+package consistent
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"reflect"
+
+	"github.com/kurin/blazer/b2"
+)
+
+const metaKey = "blazer-meta-key-no-touchie"
+
+var (
+	errUpdateConflict = errors.New("update conflict")
+	errNotInGroup     = errors.New("not in group")
+)
+
+// NewGroup creates a new consistent Group for the given bucket.
+func NewGroup(bucket *b2.Bucket, name string) *Group {
+	return &Group{
+		name: name,
+		b:    bucket,
+	}
+}
+
+// Group represents a collection of B2 objects that can be modified in a
+// consistent way.  Objects in the same group contend with each other for
+// updates, but there can only be so many (maximum of 10; fewer if there are
+// other bucket attributes set) groups in a given bucket.
+type Group struct {
+	name string
+	b    *b2.Bucket
+	ba   *b2.BucketAttrs
+}
+
+// Operate calls f with the contents of the group object given by name, and
+// updates that object with the output of f if f returns no error.  Operate
+// guarantees that no other callers have modified the contents of name in the
+// meantime (as long as all other callers are using this package).  It may call
+// f any number of times and, as a result, the potential data transfer is
+// unbounded.  Callers should have f fail after a given number of attempts if
+// this is unacceptable.
+//
+// The io.Reader that f returns is guaranteed to be read until at least the
+// first error.  Callers must ensure that this is sufficient for the reader to
+// clean up after itself.
+func (g *Group) OperateStream(ctx context.Context, name string, f func(io.Reader) (io.Reader, error)) error {
+	for {
+		r, err := g.NewReader(ctx, name)
+		if err != nil && err != errNotInGroup {
+			return err
+		}
+		out, err := f(r)
+		r.Close()
+		if err != nil {
+			return err
+		}
+		defer io.Copy(ioutil.Discard, out) // ensure the reader is read
+		w, err := g.NewWriter(ctx, r.Key, name)
+		if err != nil {
+			return err
+		}
+		if _, err := io.Copy(w, out); err != nil {
+			return err
+		}
+		if err := w.Close(); err != nil {
+			if err == errUpdateConflict {
+				continue
+			}
+			return err
+		}
+		return nil
+	}
+}
+
+// Operate uses OperateStream to act on byte slices.
+func (g *Group) Operate(ctx context.Context, name string, f func([]byte) ([]byte, error)) error {
+	return g.OperateStream(ctx, name, func(r io.Reader) (io.Reader, error) {
+		b, err := ioutil.ReadAll(r)
+		if b2.IsNotExist(err) {
+			b = nil
+			err = nil
+		}
+		if err != nil {
+			return nil, err
+		}
+		bs, err := f(b)
+		if err != nil {
+			return nil, err
+		}
+		return bytes.NewReader(bs), nil
+	})
+}
+
+// OperateJSON is a convenience function for transforming JSON data in B2 in a
+// consistent way.  Callers should pass a function f which accepts a pointer to
+// a struct of a given type and transforms it into another struct (ideally but
+// not necessarily of the same type).  Callers should also pass an example
+// struct, t, or a pointer to it, that is the same type.  t will not be
+// altered.  If there is no existing file, f will be called with an pointer to
+// an empty struct of type t.  Otherwise, it will be called with a pointer to a
+// struct filled out with the given JSON.
+func (g *Group) OperateJSON(ctx context.Context, name string, t interface{}, f func(interface{}) (interface{}, error)) error {
+	jsonType := reflect.TypeOf(t)
+	for jsonType.Kind() == reflect.Ptr {
+		jsonType = jsonType.Elem()
+	}
+	return g.OperateStream(ctx, name, func(r io.Reader) (io.Reader, error) {
+		in := reflect.New(jsonType).Interface()
+		if err := json.NewDecoder(r).Decode(in); err != nil && err != io.EOF && !b2.IsNotExist(err) {
+			return nil, err
+		}
+		out, err := f(in)
+		if err != nil {
+			return nil, err
+		}
+		pr, pw := io.Pipe()
+		go func() { pw.CloseWithError(json.NewEncoder(pw).Encode(out)) }()
+		return closeAfterReading{rc: pr}, nil
+	})
+}
+
+// closeAfterReading closes the underlying reader on the first non-nil error
+type closeAfterReading struct {
+	rc io.ReadCloser
+}
+
+func (car closeAfterReading) Read(p []byte) (int, error) {
+	n, err := car.rc.Read(p)
+	if err != nil {
+		car.rc.Close()
+	}
+	return n, err
+}
+
+// Writer is an io.ReadCloser.
+type Writer struct {
+	ctx    context.Context
+	wc     io.WriteCloser
+	name   string
+	suffix string
+	key    string
+	g      *Group
+}
+
+// Write implements io.Write.
+func (w Writer) Write(p []byte) (int, error) { return w.wc.Write(p) }
+
+// Close writes any remaining data into B2 and updates the group to reflect the
+// contents of the new object.  If the group object has been modified, Close()
+// will fail.
+func (w Writer) Close() error {
+	if err := w.wc.Close(); err != nil {
+		return err
+	}
+	// TODO: maybe see if you can cut down on calls to info()
+	for {
+		ci, err := w.g.info(w.ctx)
+		if err != nil {
+			// Replacement failed; delete the new version.
+			w.g.b.Object(w.name + "/" + w.suffix).Delete(w.ctx)
+			return err
+		}
+		old, ok := ci.Locations[w.name]
+		if ok && old != w.key {
+			w.g.b.Object(w.name + "/" + w.suffix).Delete(w.ctx)
+			return errUpdateConflict
+		}
+		ci.Locations[w.name] = w.suffix
+		if err := w.g.save(w.ctx, ci); err != nil {
+			if err == errUpdateConflict {
+				continue
+			}
+			w.g.b.Object(w.name + "/" + w.suffix).Delete(w.ctx)
+			return err
+		}
+		// Replacement successful; delete the old version.
+		w.g.b.Object(w.name + "/" + w.key).Delete(w.ctx)
+		return nil
+	}
+}
+
+// Reader is an io.ReadCloser.  Key must be passed to NewWriter.
+type Reader struct {
+	r   io.ReadCloser
+	Key string
+}
+
+func (r Reader) Read(p []byte) (int, error) {
+	if r.r == nil {
+		return 0, io.EOF
+	}
+	return r.r.Read(p)
+}
+
+func (r Reader) Close() error {
+	if r.r == nil {
+		return nil
+	}
+	return r.r.Close()
+}
+
+// NewWriter creates a Writer and prepares it to be updated.  The key argument
+// should come from the Key field of a Reader; if Writer.Close() returns with
+// no error, then the underlying group object was successfully updated from the
+// data available from the Reader with no intervening writes.  New objects can
+// be created with an empty key.
+func (g *Group) NewWriter(ctx context.Context, key, name string) (Writer, error) {
+	suffix, err := random()
+	if err != nil {
+		return Writer{}, err
+	}
+	return Writer{
+		ctx:    ctx,
+		wc:     g.b.Object(name + "/" + suffix).NewWriter(ctx),
+		name:   name,
+		suffix: suffix,
+		key:    key,
+		g:      g,
+	}, nil
+}
+
+// NewReader creates a Reader with the current version of the object, as well
+// as that object's update key.
+func (g *Group) NewReader(ctx context.Context, name string) (Reader, error) {
+	ci, err := g.info(ctx)
+	if err != nil {
+		return Reader{}, err
+	}
+	suffix, ok := ci.Locations[name]
+	if !ok {
+		return Reader{}, errNotInGroup
+	}
+	return Reader{
+		r:   g.b.Object(name + "/" + suffix).NewReader(ctx),
+		Key: suffix,
+	}, nil
+}
+
+func (g *Group) info(ctx context.Context) (*consistentInfo, error) {
+	attrs, err := g.b.Attrs(ctx)
+	if err != nil {
+		return nil, err
+	}
+	g.ba = attrs
+	imap := attrs.Info
+	if imap == nil {
+		return nil, nil
+	}
+	enc, ok := imap[metaKey+"-"+g.name]
+	if !ok {
+		return &consistentInfo{
+			Version:   1,
+			Locations: make(map[string]string),
+		}, nil
+	}
+	b, err := base64.StdEncoding.DecodeString(enc)
+	if err != nil {
+		return nil, err
+	}
+	ci := &consistentInfo{}
+	if err := json.Unmarshal(b, ci); err != nil {
+		return nil, err
+	}
+	if ci.Locations == nil {
+		ci.Locations = make(map[string]string)
+	}
+	return ci, nil
+}
+
+func (g *Group) save(ctx context.Context, ci *consistentInfo) error {
+	ci.Serial++
+	b, err := json.Marshal(ci)
+	if err != nil {
+		return err
+	}
+	s := base64.StdEncoding.EncodeToString(b)
+
+	for {
+		oldAI, err := g.info(ctx)
+		if err != nil {
+			return err
+		}
+		if oldAI.Serial != ci.Serial-1 {
+			return errUpdateConflict
+		}
+		if g.ba.Info == nil {
+			g.ba.Info = make(map[string]string)
+		}
+		g.ba.Info[metaKey+"-"+g.name] = s
+		err = g.b.Update(ctx, g.ba)
+		if err == nil {
+			return nil
+		}
+		if !b2.IsUpdateConflict(err) {
+			return err
+		}
+		// Bucket update conflict; try again.
+	}
+}
+
+// List returns a list of all the group objects.
+func (g *Group) List(ctx context.Context) ([]string, error) {
+	ci, err := g.info(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var l []string
+	for name := range ci.Locations {
+		l = append(l, name)
+	}
+	return l, nil
+}
+
+type consistentInfo struct {
+	Version int
+
+	// Serial is incremented for every version saved.  If we ensure that
+	// current.Serial = 1 + previous.Serial, and that the bucket metadata is
+	// updated cleanly, then we know that the version we saved is the direct
+	// successor to the version we had.  If the bucket metadata doesn't update
+	// cleanly, but the serial relation holds true for the new AI struct, then we
+	// can retry without bothering the user.  However, if the serial relation no
+	// longer holds true, it means someone else has updated AI and we have to ask
+	// the user to redo everything they've done.
+	//
+	// However, it is still necessary for higher level constructs to confirm that
+	// the serial number they expect is good.  The writer does this, for example,
+	// but comparing the "key" of the file it is replacing.
+	Serial    int
+	Locations map[string]string
+}
+
+func random() (string, error) {
+	b := make([]byte, 20)
+	if _, err := rand.Read(b); err != nil {
+		return "", err
+	}
+	return fmt.Sprintf("%x", b), nil
+}

vendor/github.com/kurin/blazer/x/consistent/consistent_test.go

@@ -0,0 +1,186 @@
+package consistent
+
+import (
+	"context"
+	"io"
+	"io/ioutil"
+	"os"
+	"strconv"
+	"sync"
+	"testing"
+
+	"github.com/kurin/blazer/b2"
+)
+
+const (
+	apiID      = "B2_ACCOUNT_ID"
+	apiKey     = "B2_SECRET_KEY"
+	bucketName = "consistobucket"
+)
+
+func TestOperationLive(t *testing.T) {
+	ctx := context.Background()
+	bucket, done := startLiveTest(ctx, t)
+	defer done()
+
+	g := NewGroup(bucket, "tester")
+	name := "some_kinda_name/thing.txt"
+
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		i := i
+		go func() {
+			var n int
+			defer wg.Done()
+			for j := 0; j < 10; j++ {
+				if err := g.Operate(ctx, name, func(b []byte) ([]byte, error) {
+					if len(b) > 0 {
+						i, err := strconv.Atoi(string(b))
+						if err != nil {
+							return nil, err
+						}
+						n = i
+					}
+					return []byte(strconv.Itoa(n + 1)), nil
+				}); err != nil {
+					t.Error(err)
+				}
+				t.Logf("thread %d: successful %d++", i, n)
+			}
+		}()
+	}
+	wg.Wait()
+
+	r, err := g.NewReader(ctx, name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+	b, err := ioutil.ReadAll(r)
+	if err != nil {
+		t.Fatal(err)
+	}
+	n, err := strconv.Atoi(string(b))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != 100 {
+		t.Errorf("result: got %d, want 10", n)
+	}
+}
+
+type jsonThing struct {
+	Boop   int `json:"boop_field"`
+	Thread int `json:"thread_id"`
+}
+
+func TestOperationJSONLive(t *testing.T) {
+	ctx := context.Background()
+	bucket, done := startLiveTest(ctx, t)
+	defer done()
+
+	g := NewGroup(bucket, "tester")
+	name := "some_kinda_json/thing.json"
+
+	var wg sync.WaitGroup
+	for i := 0; i < 4; i++ {
+		wg.Add(1)
+		i := i
+		go func() {
+			var n int
+			defer wg.Done()
+			for j := 0; j < 4; j++ {
+				// Pass both a struct and a pointer to a struct.
+				var face interface{}
+				face = jsonThing{}
+				if j%2 == 0 {
+					face = &jsonThing{}
+				}
+				if err := g.OperateJSON(ctx, name, face, func(j interface{}) (interface{}, error) {
+					jt := j.(*jsonThing)
+					n = jt.Boop
+					return &jsonThing{
+						Boop:   jt.Boop + 1,
+						Thread: i,
+					}, nil
+				}); err != nil {
+					t.Error(err)
+				}
+				t.Logf("thread %d: successful %d++", i, n)
+			}
+		}()
+	}
+	wg.Wait()
+
+	if err := g.OperateJSON(ctx, name, &jsonThing{}, func(i interface{}) (interface{}, error) {
+		jt := i.(*jsonThing)
+		if jt.Boop != 16 {
+			t.Errorf("got %d boops; want 16", jt.Boop)
+		}
+		return nil, nil
+	}); err != nil {
+		t.Error(err)
+	}
+}
+
+func startLiveTest(ctx context.Context, t *testing.T) (*b2.Bucket, func()) {
+	id := os.Getenv(apiID)
+	key := os.Getenv(apiKey)
+	if id == "" || key == "" {
+		t.Skipf("B2_ACCOUNT_ID or B2_SECRET_KEY unset; skipping integration tests")
+		return nil, nil
+	}
+	client, err := b2.NewClient(ctx, id, key)
+	if err != nil {
+		t.Fatal(err)
+		return nil, nil
+	}
+	bucket, err := client.NewBucket(ctx, id+"-"+bucketName, nil)
+	if err != nil {
+		t.Fatal(err)
+		return nil, nil
+	}
+	f := func() {
+		for c := range listObjects(ctx, bucket.ListObjects) {
+			if c.err != nil {
+				continue
+			}
+			if err := c.o.Delete(ctx); err != nil {
+				t.Error(err)
+			}
+		}
+		if err := bucket.Delete(ctx); err != nil && !b2.IsNotExist(err) {
+			t.Error(err)
+		}
+	}
+	return bucket, f
+}
+
+func listObjects(ctx context.Context, f func(context.Context, int, *b2.Cursor) ([]*b2.Object, *b2.Cursor, error)) <-chan object {
+	ch := make(chan object)
+	go func() {
+		defer close(ch)
+		var cur *b2.Cursor
+		for {
+			objs, c, err := f(ctx, 100, cur)
+			if err != nil && err != io.EOF {
+				ch <- object{err: err}
+				return
+			}
+			for _, o := range objs {
+				ch <- object{o: o}
+			}
+			if err == io.EOF {
+				return
+			}
+			cur = c
+		}
+	}()
+	return ch
+}
+
+type object struct {
+	o   *b2.Object
+	err error
+}