Merge branch 'pr/28'
[arvados.git] / services / keepstore / handlers.go
index d355e925d97a2c2c430c1d304e49286d2f19b592..95af1b48707c6b189982dc18762cb517769bd117 100644 (file)
@@ -8,7 +8,6 @@ package main
 // StatusHandler   (GET /status.json)
 
 import (
-       "bytes"
        "container/list"
        "crypto/md5"
        "encoding/json"
@@ -21,8 +20,7 @@ import (
        "regexp"
        "runtime"
        "strconv"
-       "strings"
-       "syscall"
+       "sync"
        "time"
 )
 
@@ -40,35 +38,19 @@ func MakeRESTRouter() *mux.Router {
 
        rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, PutBlockHandler).Methods("PUT")
        rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, DeleteHandler).Methods("DELETE")
-
-       // For IndexHandler we support:
-       //   /index           - returns all locators
-       //   /index/{prefix}  - returns all locators that begin with {prefix}
-       //      {prefix} is a string of hexadecimal digits between 0 and 32 digits.
-       //      If {prefix} is the empty string, return an index of all locators
-       //      (so /index and /index/ behave identically)
-       //      A client may supply a full 32-digit locator string, in which
-       //      case the server will return an index with either zero or one
-       //      entries. This usage allows a client to check whether a block is
-       //      present, and its size and upload time, without retrieving the
-       //      entire block.
-       //
+       // List all blocks stored here. Privileged client only.
        rest.HandleFunc(`/index`, IndexHandler).Methods("GET", "HEAD")
-       rest.HandleFunc(
-               `/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler).Methods("GET", "HEAD")
+       // List blocks stored here whose hash has the given prefix.
+       // Privileged client only.
+       rest.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler).Methods("GET", "HEAD")
+
+       // List volumes: path, device number, bytes used/avail.
        rest.HandleFunc(`/status.json`, StatusHandler).Methods("GET", "HEAD")
 
-       // The PullHandler and TrashHandler process "PUT /pull" and "PUT
-       // /trash" requests from Data Manager.  These requests instruct
-       // Keep to replicate or delete blocks; see
-       // https://arvados.org/projects/arvados/wiki/Keep_Design_Doc
-       // for more details.
-       //
-       // Each handler parses the JSON list of block management requests
-       // in the message body, and replaces any existing pull queue or
-       // trash queue with their contentes.
-       //
+       // Replace the current pull queue.
        rest.HandleFunc(`/pull`, PullHandler).Methods("PUT")
+
+       // Replace the current trash queue.
        rest.HandleFunc(`/trash`, TrashHandler).Methods("PUT")
 
        // Any request which does not match any of these routes gets
@@ -78,82 +60,37 @@ func MakeRESTRouter() *mux.Router {
        return rest
 }
 
+// BadRequestHandler is a HandleFunc to address bad requests.
 func BadRequestHandler(w http.ResponseWriter, r *http.Request) {
        http.Error(w, BadRequestError.Error(), BadRequestError.HTTPCode)
 }
 
+// GetBlockHandler is a HandleFunc to address Get block requests.
 func GetBlockHandler(resp http.ResponseWriter, req *http.Request) {
-       hash := mux.Vars(req)["hash"]
-
-       hints := mux.Vars(req)["hints"]
-
-       // Parse the locator string and hints from the request.
-       // TODO(twp): implement a Locator type.
-       var signature, timestamp string
-       if hints != "" {
-               signature_pat, _ := regexp.Compile("^A([[:xdigit:]]+)@([[:xdigit:]]{8})$")
-               for _, hint := range strings.Split(hints, "+") {
-                       if match, _ := regexp.MatchString("^[[:digit:]]+$", hint); match {
-                               // Server ignores size hints
-                       } else if m := signature_pat.FindStringSubmatch(hint); m != nil {
-                               signature = m[1]
-                               timestamp = m[2]
-                       } else if match, _ := regexp.MatchString("^[[:upper:]]", hint); match {
-                               // Any unknown hint that starts with an uppercase letter is
-                               // presumed to be valid and ignored, to permit forward compatibility.
-                       } else {
-                               // Unknown format; not a valid locator.
-                               http.Error(resp, BadRequestError.Error(), BadRequestError.HTTPCode)
-                               return
-                       }
-               }
-       }
-
-       // If permission checking is in effect, verify this
-       // request's permission signature.
-       if enforce_permissions {
-               if signature == "" || timestamp == "" {
-                       http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode)
+       if enforcePermissions {
+               locator := req.URL.Path[1:] // strip leading slash
+               if err := VerifySignature(locator, GetApiToken(req)); err != nil {
+                       http.Error(resp, err.Error(), err.(*KeepError).HTTPCode)
                        return
-               } else if IsExpired(timestamp) {
-                       http.Error(resp, ExpiredError.Error(), ExpiredError.HTTPCode)
-                       return
-               } else {
-                       req_locator := req.URL.Path[1:] // strip leading slash
-                       if !VerifySignature(req_locator, GetApiToken(req)) {
-                               http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode)
-                               return
-                       }
                }
        }
 
-       block, err := GetBlock(hash, false)
-
-       // Garbage collect after each GET. Fixes #2865.
-       // TODO(twp): review Keep memory usage and see if there's
-       // a better way to do this than blindly garbage collecting
-       // after every block.
-       defer runtime.GC()
-
+       block, err := GetBlock(mux.Vars(req)["hash"])
        if err != nil {
                // This type assertion is safe because the only errors
                // GetBlock can return are DiskHashError or NotFoundError.
                http.Error(resp, err.Error(), err.(*KeepError).HTTPCode)
                return
        }
+       defer bufs.Put(block)
 
-       resp.Header().Set("Content-Length", fmt.Sprintf("%d", len(block)))
-
-       _, err = resp.Write(block)
-
-       return
+       resp.Header().Set("Content-Length", strconv.Itoa(len(block)))
+       resp.Header().Set("Content-Type", "application/octet-stream")
+       resp.Write(block)
 }
 
+// PutBlockHandler is a HandleFunc to address Put block requests.
 func PutBlockHandler(resp http.ResponseWriter, req *http.Request) {
-       // Garbage collect after each PUT. Fixes #2865.
-       // See also GetBlockHandler.
-       defer runtime.GC()
-
        hash := mux.Vars(req)["hash"]
 
        // Detect as many error conditions as possible before reading
@@ -165,7 +102,7 @@ func PutBlockHandler(resp http.ResponseWriter, req *http.Request) {
                return
        }
 
-       if req.ContentLength > BLOCKSIZE {
+       if req.ContentLength > BlockSize {
                http.Error(resp, TooLongError.Error(), TooLongError.HTTPCode)
                return
        }
@@ -175,17 +112,17 @@ func PutBlockHandler(resp http.ResponseWriter, req *http.Request) {
                return
        }
 
-       buf := make([]byte, req.ContentLength)
-       nread, err := io.ReadFull(req.Body, buf)
+       buf := bufs.Get(int(req.ContentLength))
+       _, err := io.ReadFull(req.Body, buf)
        if err != nil {
                http.Error(resp, err.Error(), 500)
-               return
-       } else if int64(nread) < req.ContentLength {
-               http.Error(resp, "request truncated", 500)
+               bufs.Put(buf)
                return
        }
 
-       err = PutBlock(buf, hash)
+       replication, err := PutBlock(buf, hash)
+       bufs.Put(buf)
+
        if err != nil {
                ke := err.(*KeepError)
                http.Error(resp, ke.Error(), ke.HTTPCode)
@@ -194,18 +131,17 @@ func PutBlockHandler(resp http.ResponseWriter, req *http.Request) {
 
        // Success; add a size hint, sign the locator if possible, and
        // return it to the client.
-       return_hash := fmt.Sprintf("%s+%d", hash, len(buf))
-       api_token := GetApiToken(req)
-       if PermissionSecret != nil && api_token != "" {
-               expiry := time.Now().Add(blob_signature_ttl)
-               return_hash = SignLocator(return_hash, api_token, expiry)
-       }
-       resp.Write([]byte(return_hash + "\n"))
+       returnHash := fmt.Sprintf("%s+%d", hash, req.ContentLength)
+       apiToken := GetApiToken(req)
+       if PermissionSecret != nil && apiToken != "" {
+               expiry := time.Now().Add(blobSignatureTTL)
+               returnHash = SignLocator(returnHash, apiToken, expiry)
+       }
+       resp.Header().Set("X-Keep-Replicas-Stored", strconv.Itoa(replication))
+       resp.Write([]byte(returnHash + "\n"))
 }
 
-// IndexHandler
-//     A HandleFunc to address /index and /index/{prefix} requests.
-//
+// IndexHandler is a HandleFunc to address /index and /index/{prefix} requests.
 func IndexHandler(resp http.ResponseWriter, req *http.Request) {
        // Reject unauthorized requests.
        if !IsDataManagerToken(GetApiToken(req)) {
@@ -215,11 +151,21 @@ func IndexHandler(resp http.ResponseWriter, req *http.Request) {
 
        prefix := mux.Vars(req)["prefix"]
 
-       var index string
        for _, vol := range KeepVM.AllReadable() {
-               index = index + vol.Index(prefix)
+               if err := vol.IndexTo(prefix, resp); err != nil {
+                       // The only errors returned by IndexTo are
+                       // write errors returned by resp.Write(),
+                       // which probably means the client has
+                       // disconnected and this error will never be
+                       // reported to the client -- but it will
+                       // appear in our own error log.
+                       http.Error(resp, err.Error(), http.StatusInternalServerError)
+                       return
+               }
        }
-       resp.Write([]byte(index))
+       // An empty line at EOF is the only way the client can be
+       // assured the entire index was received.
+       resp.Write([]byte{'\n'})
 }
 
 // StatusHandler
@@ -233,68 +179,70 @@ func IndexHandler(resp http.ResponseWriter, req *http.Request) {
 //            * device_num (an integer identifying the underlying filesystem)
 //            * bytes_free
 //            * bytes_used
-//
-type VolumeStatus struct {
-       MountPoint string `json:"mount_point"`
-       DeviceNum  uint64 `json:"device_num"`
-       BytesFree  uint64 `json:"bytes_free"`
-       BytesUsed  uint64 `json:"bytes_used"`
+
+// PoolStatus struct
+type PoolStatus struct {
+       Alloc uint64 `json:"BytesAllocated"`
+       Cap   int    `json:"BuffersMax"`
+       Len   int    `json:"BuffersInUse"`
 }
 
+// NodeStatus struct
 type NodeStatus struct {
-       Volumes []*VolumeStatus `json:"volumes"`
+       Volumes    []*VolumeStatus `json:"volumes"`
+       BufferPool PoolStatus
+       PullQueue  WorkQueueStatus
+       TrashQueue WorkQueueStatus
+       Memory     runtime.MemStats
 }
 
+var st NodeStatus
+var stLock sync.Mutex
+
+// StatusHandler addresses /status.json requests.
 func StatusHandler(resp http.ResponseWriter, req *http.Request) {
-       st := GetNodeStatus()
-       if jstat, err := json.Marshal(st); err == nil {
+       stLock.Lock()
+       readNodeStatus(&st)
+       jstat, err := json.Marshal(&st)
+       stLock.Unlock()
+       if err == nil {
                resp.Write(jstat)
        } else {
-               log.Printf("json.Marshal: %s\n", err)
-               log.Printf("NodeStatus = %v\n", st)
+               log.Printf("json.Marshal: %s", err)
+               log.Printf("NodeStatus = %v", &st)
                http.Error(resp, err.Error(), 500)
        }
 }
 
-// GetNodeStatus
-//     Returns a NodeStatus struct describing this Keep
-//     node's current status.
-//
-func GetNodeStatus() *NodeStatus {
-       st := new(NodeStatus)
-
-       st.Volumes = make([]*VolumeStatus, len(KeepVM.AllReadable()))
-       for i, vol := range KeepVM.AllReadable() {
-               st.Volumes[i] = vol.Status()
+// populate the given NodeStatus struct with current values.
+func readNodeStatus(st *NodeStatus) {
+       vols := KeepVM.AllReadable()
+       if cap(st.Volumes) < len(vols) {
+               st.Volumes = make([]*VolumeStatus, len(vols))
        }
-       return st
+       st.Volumes = st.Volumes[:0]
+       for _, vol := range vols {
+               if s := vol.Status(); s != nil {
+                       st.Volumes = append(st.Volumes, s)
+               }
+       }
+       st.BufferPool.Alloc = bufs.Alloc()
+       st.BufferPool.Cap = bufs.Cap()
+       st.BufferPool.Len = bufs.Len()
+       st.PullQueue = getWorkQueueStatus(pullq)
+       st.TrashQueue = getWorkQueueStatus(trashq)
+       runtime.ReadMemStats(&st.Memory)
 }
 
-// GetVolumeStatus
-//     Returns a VolumeStatus describing the requested volume.
-//
-func GetVolumeStatus(volume string) *VolumeStatus {
-       var fs syscall.Statfs_t
-       var devnum uint64
-
-       if fi, err := os.Stat(volume); err == nil {
-               devnum = fi.Sys().(*syscall.Stat_t).Dev
-       } else {
-               log.Printf("GetVolumeStatus: os.Stat: %s\n", err)
-               return nil
+// return a WorkQueueStatus for the given queue. If q is nil (which
+// should never happen except in test suites), return a zero status
+// value instead of crashing.
+func getWorkQueueStatus(q *WorkQueue) WorkQueueStatus {
+       if q == nil {
+               // This should only happen during tests.
+               return WorkQueueStatus{}
        }
-
-       err := syscall.Statfs(volume, &fs)
-       if err != nil {
-               log.Printf("GetVolumeStatus: statfs: %s\n", err)
-               return nil
-       }
-       // These calculations match the way df calculates disk usage:
-       // "free" space is measured by fs.Bavail, but "used" space
-       // uses fs.Blocks - fs.Bfree.
-       free := fs.Bavail * uint64(fs.Bsize)
-       used := (fs.Blocks - fs.Bfree) * uint64(fs.Bsize)
-       return &VolumeStatus{volume, devnum, free, used}
+       return q.Status()
 }
 
 // DeleteHandler processes DELETE requests.
@@ -334,7 +282,7 @@ func DeleteHandler(resp http.ResponseWriter, req *http.Request) {
                return
        }
 
-       if never_delete {
+       if neverDelete {
                http.Error(resp, MethodDisabledError.Error(), MethodDisabledError.HTTPCode)
                return
        }
@@ -371,7 +319,7 @@ func DeleteHandler(resp http.ResponseWriter, req *http.Request) {
                if body, err := json.Marshal(result); err == nil {
                        resp.Write(body)
                } else {
-                       log.Printf("json.Marshal: %s (result = %v)\n", err, result)
+                       log.Printf("json.Marshal: %s (result = %v)", err, result)
                        http.Error(resp, err.Error(), 500)
                }
        }
@@ -410,11 +358,13 @@ func DeleteHandler(resp http.ResponseWriter, req *http.Request) {
    If the JSON unmarshalling fails, return 400 Bad Request.
 */
 
+// PullRequest consists of a block locator and an ordered list of servers
 type PullRequest struct {
        Locator string   `json:"locator"`
        Servers []string `json:"servers"`
 }
 
+// PullHandler processes "PUT /pull" requests for the data manager.
 func PullHandler(resp http.ResponseWriter, req *http.Request) {
        // Reject unauthorized requests.
        if !IsDataManagerToken(GetApiToken(req)) {
@@ -426,7 +376,7 @@ func PullHandler(resp http.ResponseWriter, req *http.Request) {
        var pr []PullRequest
        r := json.NewDecoder(req.Body)
        if err := r.Decode(&pr); err != nil {
-               http.Error(resp, BadRequestError.Error(), BadRequestError.HTTPCode)
+               http.Error(resp, err.Error(), BadRequestError.HTTPCode)
                return
        }
 
@@ -444,11 +394,13 @@ func PullHandler(resp http.ResponseWriter, req *http.Request) {
        pullq.ReplaceQueue(plist)
 }
 
+// TrashRequest consists of a block locator and it's Mtime
 type TrashRequest struct {
        Locator    string `json:"locator"`
        BlockMtime int64  `json:"block_mtime"`
 }
 
+// TrashHandler processes /trash requests.
 func TrashHandler(resp http.ResponseWriter, req *http.Request) {
        // Reject unauthorized requests.
        if !IsDataManagerToken(GetApiToken(req)) {
@@ -460,7 +412,7 @@ func TrashHandler(resp http.ResponseWriter, req *http.Request) {
        var trash []TrashRequest
        r := json.NewDecoder(req.Body)
        if err := r.Decode(&trash); err != nil {
-               http.Error(resp, BadRequestError.Error(), BadRequestError.HTTPCode)
+               http.Error(resp, err.Error(), BadRequestError.HTTPCode)
                return
        }
 
@@ -489,12 +441,9 @@ func TrashHandler(resp http.ResponseWriter, req *http.Request) {
 // should be the only part of the code that cares about which volume a
 // block is stored on, so it should be responsible for figuring out
 // which volume to check for fetching blocks, storing blocks, etc.
-
 // ==============================
-// GetBlock fetches and returns the block identified by "hash".  If
-// the update_timestamp argument is true, GetBlock also updates the
-// block's file modification time (for the sake of PutBlock, which
-// must update the file's timestamp when the block already exists).
+
+// GetBlock fetches and returns the block identified by "hash".
 //
 // On success, GetBlock returns a byte slice with the block data, and
 // a nil error.
@@ -504,23 +453,11 @@ func TrashHandler(resp http.ResponseWriter, req *http.Request) {
 // If the block found does not have the correct MD5 hash, returns
 // DiskHashError.
 //
-
-func GetBlock(hash string, update_timestamp bool) ([]byte, error) {
+func GetBlock(hash string) ([]byte, error) {
        // Attempt to read the requested hash from a keep volume.
-       error_to_caller := NotFoundError
-
-       var vols []Volume
-       if update_timestamp {
-               // Pointless to find the block on an unwritable volume
-               // because Touch() will fail -- this is as good as
-               // "not found" for purposes of callers who need to
-               // update_timestamp.
-               vols = KeepVM.AllWritable()
-       } else {
-               vols = KeepVM.AllReadable()
-       }
+       errorToCaller := NotFoundError
 
-       for _, vol := range vols {
+       for _, vol := range KeepVM.AllReadable() {
                buf, err := vol.Get(hash)
                if err != nil {
                        // IsNotExist is an expected error and may be
@@ -529,7 +466,7 @@ func GetBlock(hash string, update_timestamp bool) ([]byte, error) {
                        // volumes. If all volumes report IsNotExist,
                        // we return a NotFoundError.
                        if !os.IsNotExist(err) {
-                               log.Printf("GetBlock: reading %s: %s\n", hash, err)
+                               log.Printf("%s: Get(%s): %s", vol, hash, err)
                        }
                        continue
                }
@@ -539,138 +476,156 @@ func GetBlock(hash string, update_timestamp bool) ([]byte, error) {
                if filehash != hash {
                        // TODO: Try harder to tell a sysadmin about
                        // this.
-                       log.Printf("%s: checksum mismatch for request %s (actual %s)\n",
+                       log.Printf("%s: checksum mismatch for request %s (actual %s)",
                                vol, hash, filehash)
-                       error_to_caller = DiskHashError
+                       errorToCaller = DiskHashError
+                       bufs.Put(buf)
                        continue
                }
-               if error_to_caller == DiskHashError {
+               if errorToCaller == DiskHashError {
                        log.Printf("%s: checksum mismatch for request %s but a good copy was found on another volume and returned",
                                vol, hash)
                }
-               if update_timestamp {
-                       if err := vol.Touch(hash); err != nil {
-                               error_to_caller = GenericError
-                               log.Printf("%s: Touch %s failed: %s",
-                                       vol, hash, error_to_caller)
-                               continue
-                       }
-               }
                return buf, nil
        }
-       return nil, error_to_caller
+       return nil, errorToCaller
 }
 
-/* PutBlock(block, hash)
-   Stores the BLOCK (identified by the content id HASH) in Keep.
-
-   The MD5 checksum of the block must be identical to the content id HASH.
-   If not, an error is returned.
-
-   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:
-
-   500 Collision
-          A different block with the same hash already exists on this
-          Keep server.
-   422 MD5Fail
-          The MD5 hash of the BLOCK does not match the argument HASH.
-   503 Full
-          There was not enough space left in any Keep volume to store
-          the object.
-   500 Fail
-          The object could not be stored for some other reason (e.g.
-          all writes failed). The text of the error message should
-          provide as much detail as possible.
-*/
-
-func PutBlock(block []byte, hash string) error {
+// PutBlock Stores the BLOCK (identified by the content id HASH) in Keep.
+//
+// PutBlock(block, hash)
+//   Stores the BLOCK (identified by the content id HASH) in Keep.
+//
+//   The MD5 checksum of the block must be identical to the content id HASH.
+//   If not, an error is returned.
+//
+//   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:
+//
+//   500 Collision
+//          A different block with the same hash already exists on this
+//          Keep server.
+//   422 MD5Fail
+//          The MD5 hash of the BLOCK does not match the argument HASH.
+//   503 Full
+//          There was not enough space left in any Keep volume to store
+//          the object.
+//   500 Fail
+//          The object could not be stored for some other reason (e.g.
+//          all writes failed). The text of the error message should
+//          provide as much detail as possible.
+//
+func PutBlock(block []byte, hash string) (int, error) {
        // Check that BLOCK's checksum matches HASH.
        blockhash := fmt.Sprintf("%x", md5.Sum(block))
        if blockhash != hash {
                log.Printf("%s: MD5 checksum %s did not match request", hash, blockhash)
-               return RequestHashError
-       }
-
-       // If we already have a block on disk under this identifier, return
-       // success (but check for MD5 collisions).  While fetching the block,
-       // update its timestamp.
-       // The only errors that GetBlock can return are DiskHashError and NotFoundError.
-       // In either case, we want to write our new (good) block to disk,
-       // so there is nothing special to do if err != nil.
-       //
-       if oldblock, err := GetBlock(hash, true); err == nil {
-               if bytes.Compare(block, oldblock) == 0 {
-                       // The block already exists; return success.
-                       return nil
-               } else {
-                       return CollisionError
-               }
+               return 0, RequestHashError
+       }
+
+       // If we already have this data, it's intact on disk, and we
+       // can update its timestamp, return success. If we have
+       // different data with the same hash, return failure.
+       if n, err := CompareAndTouch(hash, block); err == nil || err == CollisionError {
+               return n, err
        }
 
        // Choose a Keep volume to write to.
        // If this volume fails, try all of the volumes in order.
        if vol := KeepVM.NextWritable(); vol != nil {
                if err := vol.Put(hash, block); err == nil {
-                       return nil // success!
+                       return vol.Replication(), nil // success!
                }
        }
 
        writables := KeepVM.AllWritable()
        if len(writables) == 0 {
                log.Print("No writable volumes.")
-               return FullError
+               return 0, FullError
        }
 
        allFull := true
        for _, vol := range writables {
                err := vol.Put(hash, block)
                if err == nil {
-                       return nil // success!
+                       return vol.Replication(), nil // success!
                }
                if err != FullError {
                        // The volume is not full but the
                        // write did not succeed.  Report the
                        // error and continue trying.
                        allFull = false
-                       log.Printf("%s: Write(%s): %s\n", vol, hash, err)
+                       log.Printf("%s: Write(%s): %s", vol, hash, err)
                }
        }
 
        if allFull {
                log.Print("All volumes are full.")
-               return FullError
-       } else {
-               // Already logged the non-full errors.
-               return GenericError
+               return 0, FullError
+       }
+       // Already logged the non-full errors.
+       return 0, GenericError
+}
+
+// CompareAndTouch returns the current replication level if one of the
+// volumes already has the given content and it successfully updates
+// the relevant block's modification time in order to protect it from
+// premature garbage collection. Otherwise, it returns a non-nil
+// error.
+func CompareAndTouch(hash string, buf []byte) (int, error) {
+       var bestErr error = NotFoundError
+       for _, vol := range KeepVM.AllWritable() {
+               if err := vol.Compare(hash, buf); err == CollisionError {
+                       // Stop if we have a block with same hash but
+                       // different content. (It will be impossible
+                       // to tell which one is wanted if we have
+                       // both, so there's no point writing it even
+                       // on a different volume.)
+                       log.Printf("%s: Compare(%s): %s", vol, hash, err)
+                       return 0, err
+               } else if os.IsNotExist(err) {
+                       // Block does not exist. This is the only
+                       // "normal" error: we don't log anything.
+                       continue
+               } else if err != nil {
+                       // Couldn't open file, data is corrupt on
+                       // disk, etc.: log this abnormal condition,
+                       // and try the next volume.
+                       log.Printf("%s: Compare(%s): %s", vol, hash, err)
+                       continue
+               }
+               if err := vol.Touch(hash); err != nil {
+                       log.Printf("%s: Touch %s failed: %s", vol, hash, err)
+                       bestErr = err
+                       continue
+               }
+               // Compare and Touch both worked --> done.
+               return vol.Replication(), nil
        }
+       return 0, bestErr
 }
 
-// IsValidLocator
-//     Return true if the specified string is a valid Keep locator.
-//     When Keep is extended to support hash types other than MD5,
-//     this should be updated to cover those as well.
+var validLocatorRe = regexp.MustCompile(`^[0-9a-f]{32}$`)
+
+// IsValidLocator returns true if the specified string is a valid Keep locator.
+//   When Keep is extended to support hash types other than MD5,
+//   this should be updated to cover those as well.
 //
 func IsValidLocator(loc string) bool {
-       match, err := regexp.MatchString(`^[0-9a-f]{32}$`, loc)
-       if err == nil {
-               return match
-       }
-       log.Printf("IsValidLocator: %s\n", err)
-       return false
+       return validLocatorRe.MatchString(loc)
 }
 
+var authRe = regexp.MustCompile(`^OAuth2\s+(.*)`)
+
 // GetApiToken returns the OAuth2 token from the Authorization
 // header of a HTTP request, or an empty string if no matching
 // token is found.
 func GetApiToken(req *http.Request) string {
        if auth, ok := req.Header["Authorization"]; ok {
-               if pat, err := regexp.Compile(`^OAuth2\s+(.*)`); err != nil {
-                       log.Println(err)
-               } else if match := pat.FindStringSubmatch(auth[0]); match != nil {
+               if match := authRe.FindStringSubmatch(auth[0]); match != nil {
                        return match[1]
                }
        }
@@ -678,36 +633,36 @@ func GetApiToken(req *http.Request) string {
 }
 
 // IsExpired returns true if the given Unix timestamp (expressed as a
-// hexadecimal string) is in the past, or if timestamp_hex cannot be
+// hexadecimal string) is in the past, or if timestampHex cannot be
 // parsed as a hexadecimal string.
-func IsExpired(timestamp_hex string) bool {
-       ts, err := strconv.ParseInt(timestamp_hex, 16, 0)
+func IsExpired(timestampHex string) bool {
+       ts, err := strconv.ParseInt(timestampHex, 16, 0)
        if err != nil {
-               log.Printf("IsExpired: %s\n", err)
+               log.Printf("IsExpired: %s", err)
                return true
        }
        return time.Unix(ts, 0).Before(time.Now())
 }
 
-// CanDelete returns true if the user identified by api_token is
+// CanDelete returns true if the user identified by apiToken is
 // allowed to delete blocks.
-func CanDelete(api_token string) bool {
-       if api_token == "" {
+func CanDelete(apiToken string) bool {
+       if apiToken == "" {
                return false
        }
        // Blocks may be deleted only when Keep has been configured with a
        // data manager.
-       if IsDataManagerToken(api_token) {
+       if IsDataManagerToken(apiToken) {
                return true
        }
-       // TODO(twp): look up api_token with the API server
+       // TODO(twp): look up apiToken with the API server
        // return true if is_admin is true and if the token
        // has unlimited scope
        return false
 }
 
-// IsDataManagerToken returns true if api_token represents the data
+// IsDataManagerToken returns true if apiToken represents the data
 // manager's token.
-func IsDataManagerToken(api_token string) bool {
-       return data_manager_token != "" && api_token == data_manager_token
+func IsDataManagerToken(apiToken string) bool {
+       return dataManagerToken != "" && apiToken == dataManagerToken
 }