PutBlock saves a block on only the first available volume (refs #2292)

Per code review: Keep stores a block on only one disk at a time.
Updated PutBlock, and PutBlock/GetBlock unit tests, appropriately.

diff --git a/services/keep/keep.go b/services/keep/keep.go
index e0883f4bf109558b06fa8c17b764204f511652d4..daa967bace5aaf4ba29072f7304e51a38a1ca689 100644 (file)
--- a/services/keep/keep.go
+++ b/services/keep/keep.go
@@ -137,7 +137,11 @@ func GetBlock(hash string) ([]byte, error) {
 
                f, err = os.Open(blockFilename)
                if err != nil {
-                       log.Printf("%s: opening %s: %s\n", vol, blockFilename, err)
+                       if !os.IsNotExist(err) {
+                               // A block is stored on only one Keep disk,
+                               // so os.IsNotExist is expected.  Report any other errors.
+                               log.Printf("%s: opening %s: %s\n", vol, blockFilename, err)
+                       }
                        continue
                }
 
@@ -175,9 +179,8 @@ func GetBlock(hash string) ([]byte, error) {
    The MD5 checksum of the block must be identical to the content id HASH.
    If not, an error is returned.
 
-   PutBlock stores the BLOCK in each of the available Keep volumes.
-   If any volume fails, an error is signaled on the back end.  A write
-   error is returned only if all volumes fail.
+   PutBlock stores the BLOCK on the first Keep volume with free space.
+   A failure code is returned to the user only if all volumes fail.
 
    On success, PutBlock returns nil.
    On failure, it returns a KeepError with one of the following codes:
@@ -201,10 +204,10 @@ func PutBlock(block []byte, hash string) error {
                return &KeepError{401, errors.New("MD5Fail")}
        }
 
-       // any_success will be set to true upon a successful block write.
-       any_success := false
        for _, vol := range KeepVolumes {
 
+               // TODO(twp): check for a full volume here before trying to write.
+
                blockDir := fmt.Sprintf("%s/%s", vol, hash[0:3])
                if err := os.MkdirAll(blockDir, 0755); err != nil {
                        log.Printf("%s: could not create directory %s: %s",
@@ -215,10 +218,11 @@ func PutBlock(block []byte, hash string) error {
                blockFilename := fmt.Sprintf("%s/%s", blockDir, hash)
                f, err := os.OpenFile(blockFilename, os.O_CREATE|os.O_WRONLY, 0644)
                if err != nil {
-                       // if the block already exists, just skip to the next volume.
+                       // if the block already exists, just return success.
+                       // TODO(twp): should we check here whether the file on disk
+                       // matches the file we were asked to store?
                        if os.IsExist(err) {
-                               any_success = true
-                               continue
+                               return nil
                        } else {
                                // Open failed for some other reason.
                                log.Printf("%s: creating %s: %s\n", vol, blockFilename, err)
@@ -228,18 +232,15 @@ func PutBlock(block []byte, hash string) error {
 
                if _, err := f.Write(block); err == nil {
                        f.Close()
-                       any_success = true
-                       continue
+                       return nil
                } else {
                        log.Printf("%s: writing to %s: %s\n", vol, blockFilename, err)
                        continue
                }
        }
 
-       if any_success {
-               return nil
-       } else {
-               log.Printf("all Keep volumes failed")
-               return &KeepError{500, errors.New("Fail")}
-       }
+       // All volumes failed; report the failure and return an error.
+       //
+       log.Printf("all Keep volumes failed")
+       return &KeepError{500, errors.New("Fail")}
 }

diff --git a/services/keep/keep_test.go b/services/keep/keep_test.go
index 027a7f4296d465ff68cd2c21203aa485aa97d5ef..bbff19d0fbac0ca272e2a5a11a160589b1f22bc1 100644 (file)
--- a/services/keep/keep_test.go
+++ b/services/keep/keep_test.go
@@ -16,37 +16,13 @@ var BAD_BLOCK = []byte("The magic words are squeamish ossifrage.")
 // GetBlock tests.
 // ========================================
 
-// TestGetBlockOK
-//     Test that a simple block read can be executed successfully.
+// TestGetBlock
+//     Test that simple block reads succeed.
 //
-func TestGetBlockOK(t *testing.T) {
+func TestGetBlock(t *testing.T) {
        defer teardown()
 
-       // Create two test Keep volumes and store a block in each of them.
-       KeepVolumes = setup(t, 2)
-
-       for _, vol := range KeepVolumes {
-               store(t, vol, TEST_HASH, TEST_BLOCK)
-       }
-
-       // Check that GetBlock returns success.
-       result, err := GetBlock(TEST_HASH)
-       if err != nil {
-               t.Errorf("GetBlock error: %s", err)
-       }
-       if fmt.Sprint(result) != fmt.Sprint(TEST_BLOCK) {
-               t.Errorf("expected %s, got %s", TEST_BLOCK, result)
-       }
-}
-
-// TestGetBlockOneKeepOK
-//     Test that block reads succeed even when the block is found only
-//     on one Keep volume.
-//
-func TestGetBlockOneKeepOK(t *testing.T) {
-       defer teardown()
-
-       // Two test Keep volumes, only the second has a block.
+       // Prepare two test Keep volumes. Our block is stored on the second volume.
        KeepVolumes = setup(t, 2)
        store(t, KeepVolumes[1], TEST_HASH, TEST_BLOCK)