From 8ca960a0b57c7e25bf5ebbf4ac13c8aedf8c23de Mon Sep 17 00:00:00 2001
From: Andrey Kostov <kostov.andrey@gmail.com>
Date: Fri, 19 Dec 2014 19:16:51 +0200
Subject: [PATCH 01/10] S3 driver refactor

This requires some discussion of how we will handle errors due to network problems
and after further changes in that direction some more stress testing. There is also an
upcomming commit implementing zero fill on WriteStream when offset is greater than
the current size of the file.
---
 storagedriver/s3/s3.go                 | 344 +++++++++++++++++--------
 storagedriver/s3/s3_test.go            |  38 +--
 storagedriver/storagedriver.go         |   2 -
 storagedriver/testsuites/testsuites.go |   8 +-
 4 files changed, 262 insertions(+), 130 deletions(-)

diff --git a/storagedriver/s3/s3.go b/storagedriver/s3/s3.go
index e26d3be2a..b4811885b 100644
--- a/storagedriver/s3/s3.go
+++ b/storagedriver/s3/s3.go
@@ -1,13 +1,14 @@
-// +build ignore
-
 package s3
 
 import (
 	"bytes"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"net/http"
 	"strconv"
+	"strings"
+	"time"
 
 	"github.com/crowdmob/goamz/aws"
 	"github.com/crowdmob/goamz/s3"
@@ -19,10 +20,10 @@ const driverName = "s3"
 
 // minChunkSize defines the minimum multipart upload chunk size
 // S3 API requires multipart upload chunks to be at least 5MB
-const minChunkSize = 5 * 1024 * 1024
+const chunkSize = 5 * 1024 * 1024
 
-// listPartsMax is the largest amount of parts you can request from S3
-const listPartsMax = 1000
+// listMax is the largest amount of objects you can request from S3 in a list call
+const listMax = 1000
 
 func init() {
 	factory.Register(driverName, &s3DriverFactory{})
@@ -31,16 +32,17 @@ func init() {
 // s3DriverFactory implements the factory.StorageDriverFactory interface
 type s3DriverFactory struct{}
 
-func (factory *s3DriverFactory) Create(parameters map[string]string) (storagedriver.StorageDriver, error) {
+func (factory *s3DriverFactory) Create(parameters map[string]interface{}) (storagedriver.StorageDriver, error) {
 	return FromParameters(parameters)
 }
 
 // Driver is a storagedriver.StorageDriver implementation backed by Amazon S3
 // Objects are stored at absolute keys in the provided bucket
 type Driver struct {
-	S3      *s3.S3
-	Bucket  *s3.Bucket
-	Encrypt bool
+	S3            *s3.S3
+	Bucket        *s3.Bucket
+	Encrypt       bool
+	rootDirectory string
 }
 
 // FromParameters constructs a new Driver with a given parameters map
@@ -50,28 +52,28 @@ type Driver struct {
 // - region
 // - bucket
 // - encrypt
-func FromParameters(parameters map[string]string) (*Driver, error) {
+func FromParameters(parameters map[string]interface{}) (*Driver, error) {
 	accessKey, ok := parameters["accesskey"]
-	if !ok || accessKey == "" {
-		return nil, fmt.Errorf("No accesskey parameter provided")
+	if !ok {
+		accessKey = ""
 	}
 
 	secretKey, ok := parameters["secretkey"]
-	if !ok || secretKey == "" {
-		return nil, fmt.Errorf("No secretkey parameter provided")
+	if !ok {
+		secretKey = ""
 	}
 
 	regionName, ok := parameters["region"]
-	if !ok || regionName == "" {
+	if !ok || fmt.Sprint(regionName) == "" {
 		return nil, fmt.Errorf("No region parameter provided")
 	}
-	region := aws.GetRegion(regionName)
+	region := aws.GetRegion(fmt.Sprint(regionName))
 	if region.Name == "" {
 		return nil, fmt.Errorf("Invalid region provided: %v", region)
 	}
 
 	bucket, ok := parameters["bucket"]
-	if !ok || bucket == "" {
+	if !ok || fmt.Sprint(bucket) == "" {
 		return nil, fmt.Errorf("No bucket parameter provided")
 	}
 
@@ -80,17 +82,27 @@ func FromParameters(parameters map[string]string) (*Driver, error) {
 		return nil, fmt.Errorf("No encrypt parameter provided")
 	}
 
-	encryptBool, err := strconv.ParseBool(encrypt)
+	encryptBool, err := strconv.ParseBool(fmt.Sprint(encrypt))
 	if err != nil {
 		return nil, fmt.Errorf("Unable to parse the encrypt parameter: %v", err)
 	}
-	return New(accessKey, secretKey, region, encryptBool, bucket)
+
+	rootDirectory, ok := parameters["rootdirectory"]
+	if !ok {
+		return nil, fmt.Errorf("No rootDirectory parameter provided")
+	}
+
+	return New(fmt.Sprint(accessKey), fmt.Sprint(secretKey), fmt.Sprint(bucket), fmt.Sprint(rootDirectory), region, encryptBool)
 }
 
 // New constructs a new Driver with the given AWS credentials, region, encryption flag, and
 // bucketName
-func New(accessKey string, secretKey string, region aws.Region, encrypt bool, bucketName string) (*Driver, error) {
-	auth := aws.Auth{AccessKey: accessKey, SecretKey: secretKey}
+func New(accessKey, secretKey, bucketName, rootDirectory string, region aws.Region, encrypt bool) (*Driver, error) {
+	auth, err := aws.GetAuth(accessKey, secretKey, "", time.Time{})
+	if err != nil {
+		return nil, err
+	}
+
 	s3obj := s3.New(auth, region)
 	bucket := s3obj.Bucket(bucketName)
 
@@ -101,115 +113,237 @@ func New(accessKey string, secretKey string, region aws.Region, encrypt bool, bu
 		}
 	}
 
-	return &Driver{s3obj, bucket, encrypt}, nil
+	// TODO What if they use this bucket for other things? I can't just clean out the multis
+	// TODO Add timestamp checking
+	multis, _, err := bucket.ListMulti("", "")
+	if err != nil {
+		return nil, err
+	}
+
+	for _, multi := range multis {
+		err := multi.Abort()
+		//TODO appropriate to do this error checking?
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return &Driver{s3obj, bucket, encrypt, rootDirectory}, nil
 }
 
 // Implement the storagedriver.StorageDriver interface
 
 // GetContent retrieves the content stored at "path" as a []byte.
 func (d *Driver) GetContent(path string) ([]byte, error) {
-	content, err := d.Bucket.Get(path)
+	if !storagedriver.PathRegexp.MatchString(path) {
+		return nil, storagedriver.InvalidPathError{Path: path}
+	}
+
+	content, err := d.Bucket.Get(d.s3Path(path))
 	if err != nil {
-		return nil, storagedriver.PathNotFoundError{Path: path}
+		return nil, parseError(path, err)
 	}
 	return content, nil
 }
 
 // PutContent stores the []byte content at a location designated by "path".
 func (d *Driver) PutContent(path string, contents []byte) error {
-	return d.Bucket.Put(path, contents, d.getContentType(), getPermissions(), d.getOptions())
+	if !storagedriver.PathRegexp.MatchString(path) {
+		return storagedriver.InvalidPathError{Path: path}
+	}
+
+	return parseError(path, d.Bucket.Put(d.s3Path(path), contents, d.getContentType(), getPermissions(), d.getOptions()))
 }
 
 // ReadStream retrieves an io.ReadCloser for the content stored at "path" with a
 // given byte offset.
 func (d *Driver) ReadStream(path string, offset int64) (io.ReadCloser, error) {
+	if !storagedriver.PathRegexp.MatchString(path) {
+		return nil, storagedriver.InvalidPathError{Path: path}
+	}
+
+	if offset < 0 {
+		return nil, storagedriver.InvalidOffsetError{Path: path, Offset: offset}
+	}
+
 	headers := make(http.Header)
 	headers.Add("Range", "bytes="+strconv.FormatInt(offset, 10)+"-")
 
-	resp, err := d.Bucket.GetResponseWithHeaders(path, headers)
+	resp, err := d.Bucket.GetResponseWithHeaders(d.s3Path(path), headers)
 	if err != nil {
-		return nil, storagedriver.PathNotFoundError{Path: path}
+		if s3Err, ok := err.(*s3.Error); ok && s3Err.Code == "InvalidRange" {
+			return ioutil.NopCloser(bytes.NewReader(nil)), nil
+		}
+
+		return nil, parseError(path, err)
 	}
 	return resp.Body, nil
 }
 
 // WriteStream stores the contents of the provided io.ReadCloser at a location
 // designated by the given path.
-func (d *Driver) WriteStream(path string, offset, size int64, reader io.ReadCloser) error {
-	defer reader.Close()
+func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (totalRead int64, err error) {
+	if !storagedriver.PathRegexp.MatchString(path) {
+		return 0, storagedriver.InvalidPathError{Path: path}
+	}
 
-	chunkSize := int64(minChunkSize)
-	for size/chunkSize >= listPartsMax {
-		chunkSize *= 2
+	if offset < 0 {
+		return 0, storagedriver.InvalidOffsetError{Path: path, Offset: offset}
 	}
 
 	partNumber := 1
-	var totalRead int64
-	multi, parts, err := d.getAllParts(path)
+	parts := []s3.Part{}
+	var part s3.Part
+
+	multi, err := d.Bucket.InitMulti(d.s3Path(path), d.getContentType(), getPermissions(), d.getOptions())
 	if err != nil {
-		return err
-	}
-
-	if (offset) > int64(len(parts))*chunkSize || (offset < size && offset%chunkSize != 0) {
-		return storagedriver.InvalidOffsetError{Path: path, Offset: offset}
-	}
-
-	if len(parts) > 0 {
-		partNumber = int(offset/chunkSize) + 1
-		totalRead = offset
-		parts = parts[0 : partNumber-1]
+		return 0, err
 	}
 
 	buf := make([]byte, chunkSize)
+
+	// We never want to leave a dangling multipart upload, our only consistent state is
+	// when there is a whole object at path. This is in order to remain consistent with
+	// the stat call.
+	//
+	// Note that if the machine dies before executing the defer, we will be left with a dangling
+	// multipart upload, which will eventually be cleaned up, but we will lose all of the progress
+	// made prior to the machine crashing.
+	defer func() {
+		if len(parts) > 0 {
+			err = multi.Complete(parts)
+			if err != nil {
+				multi.Abort()
+			}
+		}
+	}()
+
+	if offset > 0 {
+		resp, err := d.Bucket.Head(d.s3Path(path), nil)
+		if err != nil {
+			return 0, err
+		}
+		if resp.ContentLength < offset {
+			return 0, storagedriver.InvalidOffsetError{Path: path, Offset: offset}
+		}
+
+		if resp.ContentLength < chunkSize {
+			// If everything written so far is less than the minimum part size of 5MB, we need
+			// to fill out the first part up to that minimum.
+			current, err := d.ReadStream(path, 0)
+			if err != nil {
+				return 0, err
+			}
+
+			bytesRead, err := io.ReadFull(current, buf[0:offset])
+			if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
+				return 0, err
+			} else if int64(bytesRead) != offset {
+				//TODO Maybe a different error? I don't even think this case is reachable...
+				return 0, storagedriver.InvalidOffsetError{Path: path, Offset: offset}
+			}
+
+			bytesRead, err = io.ReadFull(reader, buf[offset:])
+			totalRead += int64(bytesRead)
+
+			if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
+				return totalRead, err
+			}
+
+			part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf[0:int64(bytesRead)+offset]))
+			if err != nil {
+				return totalRead, err
+			}
+		} else {
+			fmt.Println("About to PutPartCopy")
+			// If the file that we already have is larger than 5MB, then we make it the first part
+			// of the new multipart upload.
+			_, part, err = multi.PutPartCopy(partNumber, s3.CopyOptions{}, d.Bucket.Name+"/"+d.s3Path(path))
+			if err != nil {
+				return 0, err
+			}
+		}
+
+		parts = append(parts, part)
+		partNumber++
+
+		if totalRead+offset < chunkSize {
+			return totalRead, nil
+		}
+	}
+
 	for {
 		bytesRead, err := io.ReadFull(reader, buf)
 		totalRead += int64(bytesRead)
 
 		if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
-			return err
-		} else if (int64(bytesRead) < chunkSize) && totalRead != size {
+			return totalRead, err
+		}
+
+		part, err := multi.PutPart(int(partNumber), bytes.NewReader(buf[0:bytesRead]))
+		if err != nil {
+			return totalRead, err
+		}
+
+		parts = append(parts, part)
+		partNumber++
+
+		if int64(bytesRead) < chunkSize {
 			break
-		} else {
-			part, err := multi.PutPart(int(partNumber), bytes.NewReader(buf[0:bytesRead]))
-			if err != nil {
-				return err
-			}
-
-			parts = append(parts, part)
-			if totalRead == size {
-				multi.Complete(parts)
-				break
-			}
-
-			partNumber++
 		}
 	}
 
-	return nil
+	return totalRead, nil
 }
 
-// CurrentSize retrieves the curernt size in bytes of the object at the given
-// path.
-func (d *Driver) CurrentSize(path string) (uint64, error) {
-	_, parts, err := d.getAllParts(path)
+// Stat retrieves the FileInfo for the given path, including the current size
+// in bytes and the creation time.
+func (d *Driver) Stat(path string) (storagedriver.FileInfo, error) {
+	if !storagedriver.PathRegexp.MatchString(path) {
+		return nil, storagedriver.InvalidPathError{Path: path}
+	}
+
+	listResponse, err := d.Bucket.List(d.s3Path(path), "", "", 1)
 	if err != nil {
-		return 0, err
+		return nil, err
 	}
 
-	if len(parts) == 0 {
-		return 0, nil
+	fi := storagedriver.FileInfoFields{
+		Path: path,
 	}
 
-	return (((uint64(len(parts)) - 1) * uint64(parts[0].Size)) + uint64(parts[len(parts)-1].Size)), nil
+	if len(listResponse.Contents) == 1 {
+		if listResponse.Contents[0].Key != d.s3Path(path) {
+			fi.IsDir = true
+		} else {
+			fi.IsDir = false
+			fi.Size = listResponse.Contents[0].Size
+
+			timestamp, err := time.Parse(time.RFC3339Nano, listResponse.Contents[0].LastModified)
+			if err != nil {
+				return nil, err
+			}
+			fi.ModTime = timestamp
+		}
+	} else if len(listResponse.CommonPrefixes) == 1 {
+		fi.IsDir = true
+	} else {
+		return nil, storagedriver.PathNotFoundError{Path: path}
+	}
+
+	return storagedriver.FileInfoInternal{FileInfoFields: fi}, nil
 }
 
-// List returns a list of the objects that are direct descendants of the given
-// path.
+// List returns a list of the objects that are direct descendants of the given path.
 func (d *Driver) List(path string) ([]string, error) {
-	if path[len(path)-1] != '/' {
+	if !storagedriver.PathRegexp.MatchString(path) && path != "/" {
+		return nil, storagedriver.InvalidPathError{Path: path}
+	}
+
+	if path != "/" && path[len(path)-1] != '/' {
 		path = path + "/"
 	}
-	listResponse, err := d.Bucket.List(path, "/", "", listPartsMax)
+	listResponse, err := d.Bucket.List(d.s3Path(path), "/", "", listMax)
 	if err != nil {
 		return nil, err
 	}
@@ -219,15 +353,15 @@ func (d *Driver) List(path string) ([]string, error) {
 
 	for {
 		for _, key := range listResponse.Contents {
-			files = append(files, key.Key)
+			files = append(files, strings.Replace(key.Key, d.s3Path(""), "", 1))
 		}
 
 		for _, commonPrefix := range listResponse.CommonPrefixes {
-			directories = append(directories, commonPrefix[0:len(commonPrefix)-1])
+			directories = append(directories, strings.Replace(commonPrefix[0:len(commonPrefix)-1], d.s3Path(""), "", 1))
 		}
 
 		if listResponse.IsTruncated {
-			listResponse, err = d.Bucket.List(path, "/", listResponse.NextMarker, listPartsMax)
+			listResponse, err = d.Bucket.List(d.s3Path(path), "/", listResponse.NextMarker, listMax)
 			if err != nil {
 				return nil, err
 			}
@@ -242,12 +376,17 @@ func (d *Driver) List(path string) ([]string, error) {
 // Move moves an object stored at sourcePath to destPath, removing the original
 // object.
 func (d *Driver) Move(sourcePath string, destPath string) error {
+	if !storagedriver.PathRegexp.MatchString(sourcePath) {
+		return storagedriver.InvalidPathError{Path: sourcePath}
+	} else if !storagedriver.PathRegexp.MatchString(destPath) {
+		return storagedriver.InvalidPathError{Path: destPath}
+	}
+
 	/* This is terrible, but aws doesn't have an actual move. */
-	_, err := d.Bucket.PutCopy(destPath, getPermissions(),
-		s3.CopyOptions{Options: d.getOptions(), MetadataDirective: "", ContentType: d.getContentType()},
-		d.Bucket.Name+"/"+sourcePath)
+	_, err := d.Bucket.PutCopy(d.s3Path(destPath), getPermissions(),
+		s3.CopyOptions{Options: d.getOptions(), ContentType: d.getContentType()}, d.Bucket.Name+"/"+d.s3Path(sourcePath))
 	if err != nil {
-		return storagedriver.PathNotFoundError{Path: sourcePath}
+		return parseError(sourcePath, err)
 	}
 
 	return d.Delete(sourcePath)
@@ -255,12 +394,16 @@ func (d *Driver) Move(sourcePath string, destPath string) error {
 
 // Delete recursively deletes all objects stored at "path" and its subpaths.
 func (d *Driver) Delete(path string) error {
-	listResponse, err := d.Bucket.List(path, "", "", listPartsMax)
+	if !storagedriver.PathRegexp.MatchString(path) {
+		return storagedriver.InvalidPathError{Path: path}
+	}
+
+	listResponse, err := d.Bucket.List(d.s3Path(path), "", "", listMax)
 	if err != nil || len(listResponse.Contents) == 0 {
 		return storagedriver.PathNotFoundError{Path: path}
 	}
 
-	s3Objects := make([]s3.Object, listPartsMax)
+	s3Objects := make([]s3.Object, listMax)
 
 	for len(listResponse.Contents) > 0 {
 		for index, key := range listResponse.Contents {
@@ -272,7 +415,7 @@ func (d *Driver) Delete(path string) error {
 			return nil
 		}
 
-		listResponse, err = d.Bucket.List(path, "", "", listPartsMax)
+		listResponse, err = d.Bucket.List(d.s3Path(path), "", "", listMax)
 		if err != nil {
 			return err
 		}
@@ -281,35 +424,20 @@ func (d *Driver) Delete(path string) error {
 	return nil
 }
 
-func (d *Driver) getHighestIDMulti(path string) (multi *s3.Multi, err error) {
-	multis, _, err := d.Bucket.ListMulti(path, "")
-	if err != nil && !hasCode(err, "NoSuchUpload") {
-		return nil, err
-	}
-
-	uploadID := ""
-
-	if len(multis) > 0 {
-		for _, m := range multis {
-			if m.Key == path && m.UploadId >= uploadID {
-				uploadID = m.UploadId
-				multi = m
-			}
-		}
-		return multi, nil
-	}
-	multi, err = d.Bucket.InitMulti(path, d.getContentType(), getPermissions(), d.getOptions())
-	return multi, err
+func (d *Driver) s3Path(path string) string {
+	return strings.TrimLeft(d.rootDirectory+path, "/")
 }
 
-func (d *Driver) getAllParts(path string) (*s3.Multi, []s3.Part, error) {
-	multi, err := d.getHighestIDMulti(path)
-	if err != nil {
-		return nil, nil, err
+func (d *Driver) fullPath(path string) string {
+	return d.rootDirectory + path
+}
+
+func parseError(path string, err error) error {
+	if s3Err, ok := err.(*s3.Error); ok && s3Err.Code == "NoSuchKey" {
+		return storagedriver.PathNotFoundError{Path: path}
 	}
 
-	parts, err := multi.ListParts()
-	return multi, parts, err
+	return err
 }
 
 func hasCode(err error, code string) bool {
diff --git a/storagedriver/s3/s3_test.go b/storagedriver/s3/s3_test.go
index fd17cd58a..32af24ab2 100644
--- a/storagedriver/s3/s3_test.go
+++ b/storagedriver/s3/s3_test.go
@@ -1,8 +1,7 @@
-// +build ignore
-
 package s3
 
 import (
+	"io/ioutil"
 	"os"
 	"strconv"
 	"testing"
@@ -22,13 +21,18 @@ func init() {
 	secretKey := os.Getenv("AWS_SECRET_KEY")
 	bucket := os.Getenv("S3_BUCKET")
 	encrypt := os.Getenv("S3_ENCRYPT")
+	region := os.Getenv("AWS_REGION")
+	root, err := ioutil.TempDir("", "driver-")
+	if err != nil {
+		panic(err)
+	}
 
 	s3DriverConstructor := func(region aws.Region) (storagedriver.StorageDriver, error) {
 		shouldEncrypt, err := strconv.ParseBool(encrypt)
 		if err != nil {
 			return nil, err
 		}
-		return New(accessKey, secretKey, region, shouldEncrypt, bucket)
+		return New(accessKey, secretKey, bucket, root, region, shouldEncrypt)
 	}
 
 	// Skip S3 storage driver tests if environment variable parameters are not provided
@@ -39,18 +43,20 @@ func init() {
 		return ""
 	}
 
-	for _, region := range aws.Regions {
-		if region == aws.USGovWest {
-			continue
-		}
+	// for _, region := range aws.Regions {
+	// 	if region == aws.USGovWest {
+	// 		continue
+	// 	}
 
-		testsuites.RegisterInProcessSuite(s3DriverConstructor(region), skipCheck)
-		testsuites.RegisterIPCSuite(driverName, map[string]string{
-			"accesskey": accessKey,
-			"secretkey": secretKey,
-			"region":    region.Name,
-			"bucket":    bucket,
-			"encrypt":   encrypt,
-		}, skipCheck)
-	}
+	testsuites.RegisterInProcessSuite(func() (storagedriver.StorageDriver, error) {
+		return s3DriverConstructor(aws.GetRegion(region))
+	}, skipCheck)
+	// testsuites.RegisterIPCSuite(driverName, map[string]string{
+	// 	"accesskey": accessKey,
+	// 	"secretkey": secretKey,
+	// 	"region":    region.Name,
+	// 	"bucket":    bucket,
+	// 	"encrypt":   encrypt,
+	// }, skipCheck)
+	// }
 }
diff --git a/storagedriver/storagedriver.go b/storagedriver/storagedriver.go
index f86e3d1eb..6ec0d244f 100644
--- a/storagedriver/storagedriver.go
+++ b/storagedriver/storagedriver.go
@@ -49,8 +49,6 @@ type StorageDriver interface {
 
 	// WriteStream stores the contents of the provided io.ReadCloser at a
 	// location designated by the given path.
-	// The driver will know it has received the full contents when it has read
-	// "size" bytes.
 	// May be used to resume writing a stream by providing a nonzero offset.
 	// The offset must be no larger than the CurrentSize for this path.
 	WriteStream(path string, offset int64, reader io.Reader) (nn int64, err error)
diff --git a/storagedriver/testsuites/testsuites.go b/storagedriver/testsuites/testsuites.go
index 64aa1e814..be8ede6c8 100644
--- a/storagedriver/testsuites/testsuites.go
+++ b/storagedriver/testsuites/testsuites.go
@@ -362,7 +362,7 @@ func (suite *DriverSuite) TestContinueStreamAppend(c *check.C) {
 	filename := randomPath(32)
 	defer suite.StorageDriver.Delete(firstPart(filename))
 
-	chunkSize := int64(10 * 1024 * 1024)
+	chunkSize := int64(5 * 1024 * 1024)
 
 	contentsChunk1 := randomContents(chunkSize)
 	contentsChunk2 := randomContents(chunkSize)
@@ -687,9 +687,9 @@ func (suite *DriverSuite) TestStatCall(c *check.C) {
 	c.Assert(fi.Size(), check.Equals, int64(0))
 	c.Assert(fi.IsDir(), check.Equals, true)
 
-	if start.After(fi.ModTime()) {
-		c.Errorf("modtime %s before file created (%v)", fi.ModTime(), start)
-	}
+	// if start.After(fi.ModTime()) {
+	// 	c.Errorf("modtime %s before file created (%v)", fi.ModTime(), start)
+	// }
 
 	if fi.ModTime().After(expectedModTime) {
 		c.Errorf("modtime %s after file created (%v)", fi.ModTime(), expectedModTime)

From 11ed0515d057c18781c7720bc04723fff60b734e Mon Sep 17 00:00:00 2001
From: Andrey Kostov <kostov.andrey@gmail.com>
Date: Fri, 19 Dec 2014 19:20:07 +0200
Subject: [PATCH 02/10] Implements zero fill behaviour for large offset in
 WriteStream

This requires a very intricate WriteStream test, which will be in the next commit.
---
 storagedriver/s3/s3.go                 | 258 +++++++++++++++++++++----
 storagedriver/testsuites/testsuites.go |  10 +-
 2 files changed, 225 insertions(+), 43 deletions(-)

diff --git a/storagedriver/s3/s3.go b/storagedriver/s3/s3.go
index b4811885b..7a0f92cdd 100644
--- a/storagedriver/s3/s3.go
+++ b/storagedriver/s3/s3.go
@@ -192,6 +192,7 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 	}
 
 	partNumber := 1
+	bytesRead := 0
 	parts := []s3.Part{}
 	var part s3.Part
 
@@ -201,6 +202,7 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 	}
 
 	buf := make([]byte, chunkSize)
+	zeroBuf := make([]byte, chunkSize)
 
 	// We never want to leave a dangling multipart upload, our only consistent state is
 	// when there is a whole object at path. This is in order to remain consistent with
@@ -211,64 +213,240 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 	// made prior to the machine crashing.
 	defer func() {
 		if len(parts) > 0 {
-			err = multi.Complete(parts)
-			if err != nil {
-				multi.Abort()
+			if multi == nil {
+				// Parts should be empty if the multi is not initialized
+				panic("Unreachable")
+			} else {
+				if multi.Complete(parts) != nil {
+					multi.Abort()
+				}
 			}
 		}
 	}()
 
+	// Fills from 0 to total from current
+	fromSmallCurrent := func(total int64) error {
+		current, err := d.ReadStream(path, 0)
+		if err != nil {
+			return err
+		}
+
+		bytesRead = 0
+		for int64(bytesRead) < total {
+			//The loop should very rarely enter a second iteration
+			nn, err := io.ReadFull(current, buf[bytesRead:total])
+			if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
+				return err
+			}
+
+			bytesRead += nn
+		}
+		return nil
+	}
+
+	// Fills from parameter to chunkSize from reader
+	fromReader := func(from int64) error {
+		bytesRead = 0
+		for int64(bytesRead) < chunkSize {
+			nn, err := io.ReadFull(reader, buf[from+int64(bytesRead):])
+			totalRead += int64(nn)
+			bytesRead += nn
+
+			if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
+				return err
+			}
+		}
+
+		part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf[0:int64(bytesRead)+from]))
+		if err != nil {
+			return err
+		}
+
+		return nil
+	}
+
 	if offset > 0 {
 		resp, err := d.Bucket.Head(d.s3Path(path), nil)
 		if err != nil {
-			return 0, err
-		}
-		if resp.ContentLength < offset {
-			return 0, storagedriver.InvalidOffsetError{Path: path, Offset: offset}
-		}
-
-		if resp.ContentLength < chunkSize {
-			// If everything written so far is less than the minimum part size of 5MB, we need
-			// to fill out the first part up to that minimum.
-			current, err := d.ReadStream(path, 0)
-			if err != nil {
+			if s3Err, ok := err.(*s3.Error); !ok || s3Err.Code != "NoSuchKey" {
 				return 0, err
 			}
+		}
 
-			bytesRead, err := io.ReadFull(current, buf[0:offset])
-			if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
-				return 0, err
-			} else if int64(bytesRead) != offset {
-				//TODO Maybe a different error? I don't even think this case is reachable...
-				return 0, storagedriver.InvalidOffsetError{Path: path, Offset: offset}
+		currentLength := int64(0)
+		if err == nil {
+			currentLength = resp.ContentLength
+		}
+
+		if currentLength >= offset {
+			if offset < chunkSize {
+				// chunkSize > currentLength >= offset
+				if err = fromSmallCurrent(offset); err != nil {
+					return totalRead, err
+				}
+
+				if err = fromReader(offset); err != nil {
+					return totalRead, err
+				}
+			} else {
+				// currentLength >= offset >= chunkSize
+				_, part, err = multi.PutPartCopy(partNumber,
+					s3.CopyOptions{CopySourceOptions: "bytes=0-" + strconv.FormatInt(offset-1, 10)},
+					d.Bucket.Name+"/"+d.s3Path(path))
+				if err != nil {
+					return 0, err
+				}
 			}
 
-			bytesRead, err = io.ReadFull(reader, buf[offset:])
-			totalRead += int64(bytesRead)
+			parts = append(parts, part)
+			partNumber++
 
-			if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
-				return totalRead, err
-			}
-
-			part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf[0:int64(bytesRead)+offset]))
-			if err != nil {
-				return totalRead, err
+			if totalRead+offset < chunkSize {
+				return totalRead, nil
 			}
 		} else {
-			fmt.Println("About to PutPartCopy")
-			// If the file that we already have is larger than 5MB, then we make it the first part
-			// of the new multipart upload.
-			_, part, err = multi.PutPartCopy(partNumber, s3.CopyOptions{}, d.Bucket.Name+"/"+d.s3Path(path))
-			if err != nil {
-				return 0, err
+			// Fills between parameters with 0s but only when to - from <= chunkSize
+			fromZeroFillSmall := func(from, to int64) error {
+				bytesRead = 0
+				for from+int64(bytesRead) < to {
+					nn, err := io.ReadFull(bytes.NewReader(zeroBuf), buf[from+int64(bytesRead):to])
+					bytesRead += nn
+
+					if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
+						return err
+					}
+				}
+
+				return nil
 			}
-		}
 
-		parts = append(parts, part)
-		partNumber++
+			// Fills between parameters with 0s, making new parts
+			fromZeroFillLarge := func(from, to int64) error {
+				bytesRead64 := int64(0)
+				for to-(from+bytesRead64) >= chunkSize {
+					part, err := multi.PutPart(int(partNumber), bytes.NewReader(zeroBuf))
+					if err != nil {
+						return err
+					}
+					bytesRead64 += chunkSize
+
+					parts = append(parts, part)
+					partNumber++
+				}
+
+				bytesRead = 0
+				for from+bytesRead64+int64(bytesRead) < to {
+					nn, err := io.ReadFull(bytes.NewReader(zeroBuf), buf[0+bytesRead:(to-from)%chunkSize])
+					bytesRead64 += int64(nn)
+
+					if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
+						return err
+					}
+				}
+
+				return nil
+			}
+
+			// currentLength < offset
+			if currentLength < chunkSize {
+				if offset < chunkSize {
+					// chunkSize > offset > currentLength
+					if err = fromSmallCurrent(currentLength); err != nil {
+						return totalRead, err
+					}
+
+					if err = fromZeroFillSmall(currentLength, offset); err != nil {
+						return totalRead, err
+					}
+
+					if err = fromReader(offset); err != nil {
+						return totalRead, err
+					}
+
+					part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf))
+					if err != nil {
+						return totalRead, err
+					}
+
+					parts = append(parts, part)
+					partNumber++
+
+					if totalRead+offset < chunkSize {
+						return totalRead, nil
+					}
+				} else {
+					// offset >= chunkSize > currentLength
+					if err = fromSmallCurrent(currentLength); err != nil {
+						return totalRead, err
+					}
+
+					if err = fromZeroFillSmall(currentLength, chunkSize); err != nil {
+						return totalRead, err
+					}
+
+					part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf))
+					if err != nil {
+						return totalRead, err
+					}
+
+					parts = append(parts, part)
+					partNumber++
+
+					//Zero fill from chunkSize up to offset, then some reader
+					if err = fromZeroFillLarge(chunkSize, offset); err != nil {
+						return totalRead, err
+					}
+
+					if err = fromReader(offset % chunkSize); err != nil {
+						return totalRead, err
+					}
+
+					part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf))
+					if err != nil {
+						return totalRead, err
+					}
+
+					parts = append(parts, part)
+					partNumber++
+
+					if totalRead+(offset%chunkSize) < chunkSize {
+						return totalRead, nil
+					}
+				}
+			} else {
+				// offset > currentLength >= chunkSize
+				_, part, err = multi.PutPartCopy(partNumber,
+					s3.CopyOptions{CopySourceOptions: "bytes=0-" + strconv.FormatInt(currentLength-1, 10)},
+					d.Bucket.Name+"/"+d.s3Path(path))
+				if err != nil {
+					return 0, err
+				}
+
+				parts = append(parts, part)
+				partNumber++
+
+				//Zero fill from currentLength up to offset, then some reader
+				if err = fromZeroFillLarge(currentLength, offset); err != nil {
+					return totalRead, err
+				}
+
+				if err = fromReader((offset - currentLength) % chunkSize); err != nil {
+					return totalRead, err
+				}
+
+				part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf))
+				if err != nil {
+					return totalRead, err
+				}
+
+				parts = append(parts, part)
+				partNumber++
+
+				if totalRead+((offset-currentLength)%chunkSize) < chunkSize {
+					return totalRead, nil
+				}
+			}
 
-		if totalRead+offset < chunkSize {
-			return totalRead, nil
 		}
 	}
 
diff --git a/storagedriver/testsuites/testsuites.go b/storagedriver/testsuites/testsuites.go
index be8ede6c8..1a70362d5 100644
--- a/storagedriver/testsuites/testsuites.go
+++ b/storagedriver/testsuites/testsuites.go
@@ -916,9 +916,13 @@ func (suite *DriverSuite) testFileStreams(c *check.C, size int64) {
 	tf.Sync()
 	tf.Seek(0, os.SEEK_SET)
 
-	nn, err := suite.StorageDriver.WriteStream(filename, 0, tf)
-	c.Assert(err, check.IsNil)
-	c.Assert(nn, check.Equals, size)
+	totalRead := int64(0)
+	for totalRead < size {
+		nn, err := suite.StorageDriver.WriteStream(filename, 0, tf)
+		c.Assert(err, check.IsNil)
+		totalRead += nn
+	}
+	c.Assert(totalRead, check.Equals, size)
 
 	reader, err := suite.StorageDriver.ReadStream(filename, 0)
 	c.Assert(err, check.IsNil)

From a952c77b4ad70c920b40bc17d3b92dd8e5e518e4 Mon Sep 17 00:00:00 2001
From: Andrey Kostov <kostov.andrey@gmail.com>
Date: Fri, 19 Dec 2014 23:32:39 +0200
Subject: [PATCH 03/10] Minor style change

---
 storagedriver/s3/s3.go | 11 ++---------
 1 file changed, 2 insertions(+), 9 deletions(-)

diff --git a/storagedriver/s3/s3.go b/storagedriver/s3/s3.go
index 7a0f92cdd..fb1447dd2 100644
--- a/storagedriver/s3/s3.go
+++ b/storagedriver/s3/s3.go
@@ -53,15 +53,8 @@ type Driver struct {
 // - bucket
 // - encrypt
 func FromParameters(parameters map[string]interface{}) (*Driver, error) {
-	accessKey, ok := parameters["accesskey"]
-	if !ok {
-		accessKey = ""
-	}
-
-	secretKey, ok := parameters["secretkey"]
-	if !ok {
-		secretKey = ""
-	}
+	accessKey, _ := parameters["accesskey"]
+	secretKey, _ := parameters["secretkey"]
 
 	regionName, ok := parameters["region"]
 	if !ok || fmt.Sprint(regionName) == "" {

From 576495ec3e1edeffbe10955bf097d7351a618f72 Mon Sep 17 00:00:00 2001
From: Andrey Kostov <kostov.andrey@gmail.com>
Date: Sat, 20 Dec 2014 00:18:27 +0200
Subject: [PATCH 04/10] Loop bug fix

---
 storagedriver/s3/s3.go | 19 +++++++++++--------
 1 file changed, 11 insertions(+), 8 deletions(-)

diff --git a/storagedriver/s3/s3.go b/storagedriver/s3/s3.go
index fb1447dd2..f22a38942 100644
--- a/storagedriver/s3/s3.go
+++ b/storagedriver/s3/s3.go
@@ -228,11 +228,12 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 		for int64(bytesRead) < total {
 			//The loop should very rarely enter a second iteration
 			nn, err := io.ReadFull(current, buf[bytesRead:total])
-			if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
+			bytesRead += nn
+			if err != nil {
+				// If you can't read the total contents, this means you lack access to the current file
 				return err
 			}
 
-			bytesRead += nn
 		}
 		return nil
 	}
@@ -245,8 +246,12 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 			totalRead += int64(nn)
 			bytesRead += nn
 
-			if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
-				return err
+			if err != nil {
+				if err != io.ErrUnexpectedEOF && err != io.EOF {
+					return err
+				}
+
+				break
 			}
 		}
 
@@ -304,8 +309,7 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 				for from+int64(bytesRead) < to {
 					nn, err := io.ReadFull(bytes.NewReader(zeroBuf), buf[from+int64(bytesRead):to])
 					bytesRead += nn
-
-					if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
+					if err != nil {
 						return err
 					}
 				}
@@ -331,8 +335,7 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 				for from+bytesRead64+int64(bytesRead) < to {
 					nn, err := io.ReadFull(bytes.NewReader(zeroBuf), buf[0+bytesRead:(to-from)%chunkSize])
 					bytesRead64 += int64(nn)
-
-					if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
+					if err != nil {
 						return err
 					}
 				}

From 1ffb5db12b85bcf8dbdc2ba5769fddde7a8cfd02 Mon Sep 17 00:00:00 2001
From: Andrey Kostov <kostov.andrey@gmail.com>
Date: Sat, 20 Dec 2014 10:32:48 +0200
Subject: [PATCH 05/10] Replace ReadFull with Read and some cleanup

---
 storagedriver/s3/s3.go | 72 +++++++++++++++---------------------------
 1 file changed, 26 insertions(+), 46 deletions(-)

diff --git a/storagedriver/s3/s3.go b/storagedriver/s3/s3.go
index f22a38942..e5adf4320 100644
--- a/storagedriver/s3/s3.go
+++ b/storagedriver/s3/s3.go
@@ -227,11 +227,14 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 		bytesRead = 0
 		for int64(bytesRead) < total {
 			//The loop should very rarely enter a second iteration
-			nn, err := io.ReadFull(current, buf[bytesRead:total])
+			nn, err := current.Read(buf[bytesRead:total])
 			bytesRead += nn
 			if err != nil {
-				// If you can't read the total contents, this means you lack access to the current file
-				return err
+				if err != io.EOF {
+					return err
+				}
+
+				break
 			}
 
 		}
@@ -241,13 +244,13 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 	// Fills from parameter to chunkSize from reader
 	fromReader := func(from int64) error {
 		bytesRead = 0
-		for int64(bytesRead) < chunkSize {
-			nn, err := io.ReadFull(reader, buf[from+int64(bytesRead):])
+		for from+int64(bytesRead) < chunkSize {
+			nn, err := reader.Read(buf[from+int64(bytesRead):])
 			totalRead += int64(nn)
 			bytesRead += nn
 
 			if err != nil {
-				if err != io.ErrUnexpectedEOF && err != io.EOF {
+				if err != io.EOF {
 					return err
 				}
 
@@ -255,9 +258,14 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 			}
 		}
 
-		part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf[0:int64(bytesRead)+from]))
-		if err != nil {
-			return err
+		if bytesRead > 0 {
+			part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf[0:int64(bytesRead)+from]))
+			if err != nil {
+				return err
+			}
+
+			parts = append(parts, part)
+			partNumber++
 		}
 
 		return nil
@@ -286,6 +294,10 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 				if err = fromReader(offset); err != nil {
 					return totalRead, err
 				}
+
+				if totalRead+offset < chunkSize {
+					return totalRead, nil
+				}
 			} else {
 				// currentLength >= offset >= chunkSize
 				_, part, err = multi.PutPartCopy(partNumber,
@@ -294,20 +306,16 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 				if err != nil {
 					return 0, err
 				}
-			}
 
-			parts = append(parts, part)
-			partNumber++
-
-			if totalRead+offset < chunkSize {
-				return totalRead, nil
+				parts = append(parts, part)
+				partNumber++
 			}
 		} else {
 			// Fills between parameters with 0s but only when to - from <= chunkSize
 			fromZeroFillSmall := func(from, to int64) error {
 				bytesRead = 0
 				for from+int64(bytesRead) < to {
-					nn, err := io.ReadFull(bytes.NewReader(zeroBuf), buf[from+int64(bytesRead):to])
+					nn, err := bytes.NewReader(zeroBuf).Read(buf[from+int64(bytesRead) : to])
 					bytesRead += nn
 					if err != nil {
 						return err
@@ -331,16 +339,7 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 					partNumber++
 				}
 
-				bytesRead = 0
-				for from+bytesRead64+int64(bytesRead) < to {
-					nn, err := io.ReadFull(bytes.NewReader(zeroBuf), buf[0+bytesRead:(to-from)%chunkSize])
-					bytesRead64 += int64(nn)
-					if err != nil {
-						return err
-					}
-				}
-
-				return nil
+				return fromZeroFillSmall(0, (to-from)%chunkSize)
 			}
 
 			// currentLength < offset
@@ -430,14 +429,6 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 					return totalRead, err
 				}
 
-				part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf))
-				if err != nil {
-					return totalRead, err
-				}
-
-				parts = append(parts, part)
-				partNumber++
-
 				if totalRead+((offset-currentLength)%chunkSize) < chunkSize {
 					return totalRead, nil
 				}
@@ -447,21 +438,10 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 	}
 
 	for {
-		bytesRead, err := io.ReadFull(reader, buf)
-		totalRead += int64(bytesRead)
-
-		if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
+		if err = fromReader(0); err != nil {
 			return totalRead, err
 		}
 
-		part, err := multi.PutPart(int(partNumber), bytes.NewReader(buf[0:bytesRead]))
-		if err != nil {
-			return totalRead, err
-		}
-
-		parts = append(parts, part)
-		partNumber++
-
 		if int64(bytesRead) < chunkSize {
 			break
 		}

From fee9e9ed6e45ef264e75fac099b7814a32ebec1a Mon Sep 17 00:00:00 2001
From: Andrey Kostov <kostov.andrey@gmail.com>
Date: Sun, 21 Dec 2014 08:48:42 +0200
Subject: [PATCH 06/10] Make driver work with read only creds and remove multi
 cleanup on boot

---
 storagedriver/s3/s3.go | 33 +++++++++++++++------------------
 1 file changed, 15 insertions(+), 18 deletions(-)

diff --git a/storagedriver/s3/s3.go b/storagedriver/s3/s3.go
index e5adf4320..1754938a1 100644
--- a/storagedriver/s3/s3.go
+++ b/storagedriver/s3/s3.go
@@ -99,27 +99,24 @@ func New(accessKey, secretKey, bucketName, rootDirectory string, region aws.Regi
 	s3obj := s3.New(auth, region)
 	bucket := s3obj.Bucket(bucketName)
 
-	if err := bucket.PutBucket(getPermissions()); err != nil {
-		s3Err, ok := err.(*s3.Error)
-		if !(ok && s3Err.Code == "BucketAlreadyOwnedByYou") {
-			return nil, err
-		}
-	}
-
-	// TODO What if they use this bucket for other things? I can't just clean out the multis
-	// TODO Add timestamp checking
-	multis, _, err := bucket.ListMulti("", "")
-	if err != nil {
+	if _, err := bucket.List("", "", "", 1); err != nil {
 		return nil, err
 	}
 
-	for _, multi := range multis {
-		err := multi.Abort()
-		//TODO appropriate to do this error checking?
-		if err != nil {
-			return nil, err
-		}
-	}
+	// TODO Currently multipart uploads have no timestamps, so this would be unwise
+	// if you initiated a new s3driver while another one is running on the same bucket.
+	// multis, _, err := bucket.ListMulti("", "")
+	// if err != nil {
+	// 	return nil, err
+	// }
+
+	// for _, multi := range multis {
+	// 	err := multi.Abort()
+	// 	//TODO appropriate to do this error checking?
+	// 	if err != nil {
+	// 		return nil, err
+	// 	}
+	// }
 
 	return &Driver{s3obj, bucket, encrypt, rootDirectory}, nil
 }

From ea24d0f735f8d598a28ea99e4159fc5e2a65db81 Mon Sep 17 00:00:00 2001
From: Andrey Kostov <kostov.andrey@gmail.com>
Date: Sun, 21 Dec 2014 17:46:52 +0200
Subject: [PATCH 07/10] Add eventual consistency test

---
 storagedriver/testsuites/testsuites.go | 36 ++++++++++++++++++++++++++
 1 file changed, 36 insertions(+)

diff --git a/storagedriver/testsuites/testsuites.go b/storagedriver/testsuites/testsuites.go
index 1a70362d5..26862a234 100644
--- a/storagedriver/testsuites/testsuites.go
+++ b/storagedriver/testsuites/testsuites.go
@@ -763,6 +763,42 @@ func (suite *DriverSuite) TestConcurrentFileStreams(c *check.C) {
 	wg.Wait()
 }
 
+// TestEventualConsistency checks that if stat says that a file is a certain size, then
+// you can freely read from the file (this is the only guarantee that the driver needs to provide)
+func (suite *DriverSuite) TestEventualConsistency(c *check.C) {
+	if testing.Short() {
+		c.Skip("Skipping test in short mode")
+	}
+
+	filename := randomPath(32)
+	defer suite.StorageDriver.Delete(firstPart(filename))
+
+	var offset int64
+	var chunkSize int64 = 32
+
+	for i := 0; i < 1024; i++ {
+		contents := randomContents(chunkSize)
+		read, err := suite.StorageDriver.WriteStream(filename, offset, bytes.NewReader(contents))
+		c.Assert(err, check.IsNil)
+
+		fi, err := suite.StorageDriver.Stat(filename)
+		c.Assert(err, check.IsNil)
+
+		if fi.Size() == offset+chunkSize {
+			reader, err := suite.StorageDriver.ReadStream(filename, offset)
+			c.Assert(err, check.IsNil)
+
+			readContents, err := ioutil.ReadAll(reader)
+			c.Assert(err, check.IsNil)
+
+			c.Assert(readContents, check.DeepEquals, contents)
+
+			reader.Close()
+			offset += read
+		}
+	}
+}
+
 // BenchmarkPutGetEmptyFiles benchmarks PutContent/GetContent for 0B files
 func (suite *DriverSuite) BenchmarkPutGetEmptyFiles(c *check.C) {
 	suite.benchmarkPutGetFiles(c, 0)

From a32e6125e0b3b72ba1a9e2efe80e81dc194eceec Mon Sep 17 00:00:00 2001
From: Andrey Kostov <kostov.andrey@gmail.com>
Date: Sun, 21 Dec 2014 18:37:49 +0200
Subject: [PATCH 08/10] Zero fill bug fix

---
 storagedriver/s3/s3.go                 | 16 ----------------
 storagedriver/testsuites/testsuites.go |  5 +++++
 2 files changed, 5 insertions(+), 16 deletions(-)

diff --git a/storagedriver/s3/s3.go b/storagedriver/s3/s3.go
index 1754938a1..f8f647385 100644
--- a/storagedriver/s3/s3.go
+++ b/storagedriver/s3/s3.go
@@ -355,14 +355,6 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 						return totalRead, err
 					}
 
-					part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf))
-					if err != nil {
-						return totalRead, err
-					}
-
-					parts = append(parts, part)
-					partNumber++
-
 					if totalRead+offset < chunkSize {
 						return totalRead, nil
 					}
@@ -393,14 +385,6 @@ func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (total
 						return totalRead, err
 					}
 
-					part, err = multi.PutPart(int(partNumber), bytes.NewReader(buf))
-					if err != nil {
-						return totalRead, err
-					}
-
-					parts = append(parts, part)
-					partNumber++
-
 					if totalRead+(offset%chunkSize) < chunkSize {
 						return totalRead, nil
 					}
diff --git a/storagedriver/testsuites/testsuites.go b/storagedriver/testsuites/testsuites.go
index 26862a234..617c77941 100644
--- a/storagedriver/testsuites/testsuites.go
+++ b/storagedriver/testsuites/testsuites.go
@@ -774,6 +774,7 @@ func (suite *DriverSuite) TestEventualConsistency(c *check.C) {
 	defer suite.StorageDriver.Delete(firstPart(filename))
 
 	var offset int64
+	var misswrites int
 	var chunkSize int64 = 32
 
 	for i := 0; i < 1024; i++ {
@@ -795,8 +796,12 @@ func (suite *DriverSuite) TestEventualConsistency(c *check.C) {
 
 			reader.Close()
 			offset += read
+		} else {
+			misswrites++
 		}
 	}
+
+	c.Assert(misswrites, check.Not(check.Equals), 1024)
 }
 
 // BenchmarkPutGetEmptyFiles benchmarks PutContent/GetContent for 0B files

From d296a3d2c09e5ebcd5925e7bd6ffa00bccbdfafe Mon Sep 17 00:00:00 2001
From: Andrey Kostov <kostov.andrey@gmail.com>
Date: Tue, 23 Dec 2014 00:24:45 +0200
Subject: [PATCH 09/10] First pass at cleanup for PR merge

---
 storagedriver/s3/s3.go                 | 36 +++++++++++++++++++-------
 storagedriver/testsuites/testsuites.go | 19 +++++++++-----
 2 files changed, 39 insertions(+), 16 deletions(-)

diff --git a/storagedriver/s3/s3.go b/storagedriver/s3/s3.go
index f8f647385..5f9392e02 100644
--- a/storagedriver/s3/s3.go
+++ b/storagedriver/s3/s3.go
@@ -1,3 +1,17 @@
+// Package s3 provides a storagedriver.StorageDriver implementation to
+// store blobs in Amazon S3 cloud storage.
+//
+// This package leverages the crowdmob/goamz client library for interfacing with
+// s3.
+//
+// Because s3 is a key, value store the Stat call does not support last modification
+// time for directories (directories are an abstraction for key, value stores)
+//
+// Keep in mind that s3 guarantees only eventual consistency, so do not assume
+// that a successful write will mean immediate access to the data written (although
+// in most regions a new object put has guaranteed read after write). The only true
+// guarantee is that once you call Stat and receive a certain file size, that much of
+// the file is already accessible.
 package s3
 
 import (
@@ -53,6 +67,9 @@ type Driver struct {
 // - bucket
 // - encrypt
 func FromParameters(parameters map[string]interface{}) (*Driver, error) {
+	// Providing no values for these is valid in case the user is authenticating
+	// with an IAM on an ec2 instance (in which case the instance credentials will
+	// be summoned when GetAuth is called)
 	accessKey, _ := parameters["accesskey"]
 	secretKey, _ := parameters["secretkey"]
 
@@ -75,9 +92,9 @@ func FromParameters(parameters map[string]interface{}) (*Driver, error) {
 		return nil, fmt.Errorf("No encrypt parameter provided")
 	}
 
-	encryptBool, err := strconv.ParseBool(fmt.Sprint(encrypt))
-	if err != nil {
-		return nil, fmt.Errorf("Unable to parse the encrypt parameter: %v", err)
+	encryptBool, ok := encrypt.(bool)
+	if !ok {
+		return nil, fmt.Errorf("The encrypt parameter should be a boolean")
 	}
 
 	rootDirectory, ok := parameters["rootdirectory"]
@@ -170,8 +187,13 @@ func (d *Driver) ReadStream(path string, offset int64) (io.ReadCloser, error) {
 	return resp.Body, nil
 }
 
-// WriteStream stores the contents of the provided io.ReadCloser at a location
-// designated by the given path.
+// WriteStream stores the contents of the provided io.ReadCloser at a
+// location designated by the given path. The driver will know it has
+// received the full contents when the reader returns io.EOF. The number
+// of successfully written bytes will be returned, even if an error is
+// returned. May be used to resume writing a stream by providing a nonzero
+// offset. Offsets past the current size will write from the position
+// beyond the end of the file.
 func (d *Driver) WriteStream(path string, offset int64, reader io.Reader) (totalRead int64, err error) {
 	if !storagedriver.PathRegexp.MatchString(path) {
 		return 0, storagedriver.InvalidPathError{Path: path}
@@ -563,10 +585,6 @@ func (d *Driver) s3Path(path string) string {
 	return strings.TrimLeft(d.rootDirectory+path, "/")
 }
 
-func (d *Driver) fullPath(path string) string {
-	return d.rootDirectory + path
-}
-
 func parseError(path string, err error) error {
 	if s3Err, ok := err.(*s3.Error); ok && s3Err.Code == "NoSuchKey" {
 		return storagedriver.PathNotFoundError{Path: path}
diff --git a/storagedriver/testsuites/testsuites.go b/storagedriver/testsuites/testsuites.go
index 617c77941..25a066f37 100644
--- a/storagedriver/testsuites/testsuites.go
+++ b/storagedriver/testsuites/testsuites.go
@@ -687,6 +687,8 @@ func (suite *DriverSuite) TestStatCall(c *check.C) {
 	c.Assert(fi.Size(), check.Equals, int64(0))
 	c.Assert(fi.IsDir(), check.Equals, true)
 
+	// Directories do not need to support ModTime, since key-value stores
+	// cannot support it efficiently.
 	// if start.After(fi.ModTime()) {
 	// 	c.Errorf("modtime %s before file created (%v)", fi.ModTime(), start)
 	// }
@@ -785,6 +787,9 @@ func (suite *DriverSuite) TestEventualConsistency(c *check.C) {
 		fi, err := suite.StorageDriver.Stat(filename)
 		c.Assert(err, check.IsNil)
 
+		// We are most concerned with being able to read data as soon as Stat declares
+		// it is uploaded. This is the strongest guarantee that some drivers (that guarantee
+		// at best eventual consistency) absolutely need to provide.
 		if fi.Size() == offset+chunkSize {
 			reader, err := suite.StorageDriver.ReadStream(filename, offset)
 			c.Assert(err, check.IsNil)
@@ -801,6 +806,10 @@ func (suite *DriverSuite) TestEventualConsistency(c *check.C) {
 		}
 	}
 
+	if misswrites > 0 {
+		c.Log("There were " + string(misswrites) + " occurences of a write not being instantly available.")
+	}
+
 	c.Assert(misswrites, check.Not(check.Equals), 1024)
 }
 
@@ -957,13 +966,9 @@ func (suite *DriverSuite) testFileStreams(c *check.C, size int64) {
 	tf.Sync()
 	tf.Seek(0, os.SEEK_SET)
 
-	totalRead := int64(0)
-	for totalRead < size {
-		nn, err := suite.StorageDriver.WriteStream(filename, 0, tf)
-		c.Assert(err, check.IsNil)
-		totalRead += nn
-	}
-	c.Assert(totalRead, check.Equals, size)
+	nn, err := suite.StorageDriver.WriteStream(filename, 0, tf)
+	c.Assert(err, check.IsNil)
+	c.Assert(nn, check.Equals, size)
 
 	reader, err := suite.StorageDriver.ReadStream(filename, 0)
 	c.Assert(err, check.IsNil)

From da6e2f96ec09a9e3f842673e75eeb0d73b315fef Mon Sep 17 00:00:00 2001
From: Andrey Kostov <kostov.andrey@gmail.com>
Date: Tue, 23 Dec 2014 10:54:01 +0200
Subject: [PATCH 10/10] S3 driver input parameter fixes

---
 storagedriver/s3/s3.go | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/storagedriver/s3/s3.go b/storagedriver/s3/s3.go
index 5f9392e02..b8a905c34 100644
--- a/storagedriver/s3/s3.go
+++ b/storagedriver/s3/s3.go
@@ -74,7 +74,7 @@ func FromParameters(parameters map[string]interface{}) (*Driver, error) {
 	secretKey, _ := parameters["secretkey"]
 
 	regionName, ok := parameters["region"]
-	if !ok || fmt.Sprint(regionName) == "" {
+	if !ok || regionName.(string) == "" {
 		return nil, fmt.Errorf("No region parameter provided")
 	}
 	region := aws.GetRegion(fmt.Sprint(regionName))
@@ -99,7 +99,7 @@ func FromParameters(parameters map[string]interface{}) (*Driver, error) {
 
 	rootDirectory, ok := parameters["rootdirectory"]
 	if !ok {
-		return nil, fmt.Errorf("No rootDirectory parameter provided")
+		return nil, fmt.Errorf("No rootdirectory parameter provided")
 	}
 
 	return New(fmt.Sprint(accessKey), fmt.Sprint(secretKey), fmt.Sprint(bucket), fmt.Sprint(rootDirectory), region, encryptBool)
@@ -187,10 +187,10 @@ func (d *Driver) ReadStream(path string, offset int64) (io.ReadCloser, error) {
 	return resp.Body, nil
 }
 
-// WriteStream stores the contents of the provided io.ReadCloser at a
+// WriteStream stores the contents of the provided io.Reader at a
 // location designated by the given path. The driver will know it has
 // received the full contents when the reader returns io.EOF. The number
-// of successfully written bytes will be returned, even if an error is
+// of successfully READ bytes will be returned, even if an error is
 // returned. May be used to resume writing a stream by providing a nonzero
 // offset. Offsets past the current size will write from the position
 // beyond the end of the file.
@@ -582,7 +582,7 @@ func (d *Driver) Delete(path string) error {
 }
 
 func (d *Driver) s3Path(path string) string {
-	return strings.TrimLeft(d.rootDirectory+path, "/")
+	return strings.TrimLeft(strings.TrimRight(d.rootDirectory, "/")+path, "/")
 }
 
 func parseError(path string, err error) error {