replaced storage.BlockExists() with storage.BlockSize() which returns additional info

internal/storagetesting/asserts.go

@@ -34,14 +34,19 @@ func AssertGetBlockNotFound(t *testing.T, s storage.Storage, block string) {
 
 // AssertBlockExists asserts that BlockExists() the specified storage block returns the correct value.
 func AssertBlockExists(t *testing.T, s storage.Storage, block string, expected bool) {
-	e, err := s.BlockExists(block)
-	if err != nil {
-		t.Errorf(errorPrefix()+"BlockExists(%v) returned error %v, expected: %v", block, err, expected)
+	_, err := s.BlockSize(block)
+	var exists bool
+	if err == nil {
+		exists = true
+	} else if err == storage.ErrBlockNotFound {
+		exists = false
+	} else {
+		t.Errorf(errorPrefix()+"BlockSize(%v) returned error: %v", block, err)
 		return
 	}
 
-	if !reflect.DeepEqual(e, expected) {
-		t.Errorf(errorPrefix()+"BlockExists(%v) returned %v, but expected %v", block, e, expected)
+	if exists != expected {
+		t.Errorf(errorPrefix()+"BlockSize(%v) returned exists=%v, but expected %v", block, exists, expected)
 	}
 }
 

internal/storagetesting/map.go

@@ -14,11 +14,15 @@ type mapStorage struct {
 	mutex sync.RWMutex
 }
 
-func (s *mapStorage) BlockExists(id string) (bool, error) {
+func (s *mapStorage) BlockSize(id string) (int64, error) {
 	s.mutex.RLock()
 	defer s.mutex.RUnlock()
-	_, ok := s.data[string(id)]
-	return ok, nil
+	d, ok := s.data[string(id)]
+	if !ok {
+		return 0, storage.ErrBlockNotFound
+	}
+
+	return int64(len(d)), nil
 }
 
 func (s *mapStorage) GetBlock(id string) ([]byte, error) {
@@ -73,7 +77,7 @@ func (s *mapStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
 			v := s.data[k]
 			ch <- storage.BlockMetadata{
 				BlockID:   string(k),
-				Length:    uint64(len(v)),
+				Length:    int64(len(v)),
 				TimeStamp: fixedTime,
 			}
 		}

internal/storagetesting/verify.go

@@ -21,7 +21,7 @@ func VerifyStorage(t *testing.T, r storage.Storage) {
 
 	// First verify that blocks don't exist.
 	for _, b := range blocks {
-		if x, err := r.BlockExists(b.blk); x || err != nil {
+		if _, err := r.BlockSize(b.blk); err != storage.ErrBlockNotFound {
 			t.Errorf("block exists or error: %v %v", b.blk, err)
 		}
 

repo/repository.go

@@ -287,15 +287,15 @@ func (r *repository) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix
 	}
 
 	// Before performing encryption, check if the block is already there.
-	ok, err := r.storage.BlockExists(objectID.StorageBlock)
-	if err != nil {
-		// Don't know whether block exists in storage.
-		return NullObjectID, err
+	blockSize, err := r.storage.BlockSize(objectID.StorageBlock)
+	if err == nil && blockSize == int64(len(data)) {
+		// Block already exists in storage, correct size, return without uploading.
+		return objectID, nil
 	}
 
-	if ok {
-		// Block already exists in storage, return without uploading.
-		return objectID, nil
+	if err != nil && err != storage.ErrBlockNotFound {
+		// Don't know whether block exists in storage.
+		return NullObjectID, err
 	}
 
 	// Encryption is requested, encrypt the block in-place.

storage/caching/cache_entry.go

@@ -11,7 +11,6 @@ import (
 const (
 	cacheEntryFormatVersion = 1
 	sizeDoesNotExists       = 0x4000000000000000
-	sizeUnknown             = 0x4000000000000001
 )
 
 type blockCacheEntry struct {
@@ -23,17 +22,11 @@ func (e *blockCacheEntry) exists() bool {
 	return e.size != sizeDoesNotExists
 }
 
-func (e *blockCacheEntry) isKnownSize() bool {
-	return e.size != sizeUnknown
-}
-
 func (e blockCacheEntry) GoString() string {
 	ts := time.Unix(e.accessTime/1000000000, e.accessTime%1000000000)
 	switch e.size {
 	case sizeDoesNotExists:
 		return fmt.Sprintf("entry[not-found;acc:%v]", ts)
-	case sizeUnknown:
-		return fmt.Sprintf("entry[exists-size-unknown;acc:%v]", ts)
 	default:
 		return fmt.Sprintf("entry[size:%v;acc:%v]", e.size, ts)
 	}

storage/caching/caching_storage.go

@@ -96,21 +96,28 @@ func (c *cachingStorage) removeCacheEntry(block string) {
 	})
 }
 
-func (c *cachingStorage) BlockExists(id string) (bool, error) {
+func (c *cachingStorage) BlockSize(id string) (int64, error) {
 	if entry, ok := c.getCacheEntry(id); ok {
-		return entry.exists(), nil
+		if entry.exists() {
+			return entry.size, nil
+		}
+
+		return 0, storage.ErrBlockNotFound
 	}
 
 	c.Lock(id)
 	defer c.Unlock(id)
 
-	exists, err := c.master.BlockExists(id)
+	l, err := c.master.BlockSize(id)
 	if err != nil {
-		return false, err
+		if err == storage.ErrBlockNotFound {
+			c.setCacheEntrySize(id, sizeDoesNotExists)
+		}
+		return 0, err
 	}
 
-	c.setCacheEntrySize(id, sizeUnknown)
-	return exists, nil
+	c.setCacheEntrySize(id, l)
+	return l, nil
 }
 
 func (c *cachingStorage) DeleteBlock(id string) error {
@@ -137,8 +144,9 @@ func (c *cachingStorage) GetBlock(id string) ([]byte, error) {
 			return nil, storage.ErrBlockNotFound
 		}
 
-		if blockCacheEntry.isKnownSize() {
-			return c.cache.GetBlock(id)
+		v, err := c.cache.GetBlock(id)
+		if err == nil {
+			return v, nil
 		}
 	}
 

storage/caching/caching_storage_test.go

@@ -131,7 +131,8 @@ func TestCache(t *testing.T) {
 	tr.assertActivityAndClear(t, "PutBlock")
 
 	storagetesting.AssertBlockExists(t, cache, "z", true)
-	tr.assertActivityAndClear(t, "BlockExists")
+	tr.assertActivityAndClear(t, "BlockSize")
+
 	storagetesting.AssertGetBlock(t, cache, "z", data1)
 	tr.assertActivityAndClear(t, "GetBlock")
 

storage/filesystem/filesystem_storage.go

@@ -28,18 +28,18 @@ type fsStorage struct {
 	Options
 }
 
-func (fs *fsStorage) BlockExists(blockID string) (bool, error) {
+func (fs *fsStorage) BlockSize(blockID string) (int64, error) {
 	_, path := fs.getShardedPathAndFilePath(blockID)
-	_, err := os.Stat(path)
+	s, err := os.Stat(path)
 	if err == nil {
-		return true, nil
+		return s.Size(), nil
 	}
 
 	if os.IsNotExist(err) {
-		return false, nil
+		return 0, storage.ErrBlockNotFound
 	}
 
-	return false, err
+	return 0, err
 }
 
 func (fs *fsStorage) GetBlock(blockID string) ([]byte, error) {
@@ -97,7 +97,7 @@ func (fs *fsStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
 					if strings.HasPrefix(string(fullID), prefixString) {
 						result <- storage.BlockMetadata{
 							BlockID:   fullID,
-							Length:    uint64(e.Size()),
+							Length:    e.Size(),
 							TimeStamp: e.ModTime(),
 						}
 					}

storage/gcs/gcs_storage.go

@@ -39,23 +39,19 @@ type gcsStorage struct {
 	objectsService *gcsclient.ObjectsService
 }
 
-func (gcs *gcsStorage) BlockExists(b string) (bool, error) {
+func (gcs *gcsStorage) BlockSize(b string) (int64, error) {
 	call := gcs.objectsService.Get(gcs.BucketName, gcs.getObjectNameString(b))
-	_, err := retry(
+	v, err := retry(
 		"Get",
 		func() (interface{}, error) {
 			return call.Do()
 		})
 
-	if err == nil {
-		return true, nil
-	}
-
 	if isGoogleAPIError(err, http.StatusNotFound) {
-		return false, nil
+		return 0, storage.ErrBlockNotFound
 	}
 
-	return false, err
+	return int64(v.(*gcsclient.Object).Size), nil
 }
 
 func (gcs *gcsStorage) GetBlock(b string) ([]byte, error) {
@@ -159,7 +155,7 @@ func (gcs *gcsStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
 				} else {
 					ch <- storage.BlockMetadata{
 						BlockID:   string(o.Name)[len(gcs.Prefix):],
-						Length:    o.Size,
+						Length:    int64(o.Size),
 						TimeStamp: t,
 					}
 				}

storage/logging/logging_storage.go

@@ -14,11 +14,11 @@ type loggingStorage struct {
 	prefix string
 }
 
-func (s *loggingStorage) BlockExists(id string) (bool, error) {
+func (s *loggingStorage) BlockSize(id string) (int64, error) {
 	t0 := time.Now()
-	result, err := s.base.BlockExists(id)
+	result, err := s.base.BlockSize(id)
 	dt := time.Since(t0)
-	s.printf(s.prefix+"BlockExists(%#v)=%#v,%#v took %v", id, result, err, dt)
+	s.printf(s.prefix+"BlockSize(%#v)=%#v,%#v took %v", id, result, err, dt)
 	return result, err
 }
 

storage/storage.go

@@ -18,9 +18,9 @@ const (
 type Storage interface {
 	io.Closer
 
+	BlockSize(id string) (int64, error)
 	PutBlock(id string, data []byte, options PutOptions) error
 	DeleteBlock(id string) error
-	BlockExists(id string) (bool, error)
 	GetBlock(id string) ([]byte, error)
 	ListBlocks(prefix string) chan (BlockMetadata)
 }
@@ -34,7 +34,7 @@ type ConnectionInfoProvider interface {
 // If Error field is set, no other field values should be assumed to be correct.
 type BlockMetadata struct {
 	BlockID   string
-	Length    uint64
+	Length    int64
 	TimeStamp time.Time
 	Error     error
 }