Merge pull request #487 from jaddr2line/cluster-message-error

Differentiate between cluster message request errors and cluster message processing errors

api.go

@@ -795,11 +795,30 @@ func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error {
 
 	// Forward the message.
 	if err := api.server.receiveMessage(msg); err != nil {
-		return errors.Wrap(err, "receiving message")
+		return MessageProcessingError{err}
 	}
 	return nil
 }
 
+// MessageProcessingError is an error indicating that a cluster message could not be processed.
+type MessageProcessingError struct {
+	Err error
+}
+
+func (err MessageProcessingError) Error() string {
+	return "processing message: " + err.Err.Error()
+}
+
+// Cause allows the error to be unwrapped.
+func (err MessageProcessingError) Cause() error {
+	return err.Err
+}
+
+// Unwrap allows the error to be unwrapped.
+func (err MessageProcessingError) Unwrap() error {
+	return err.Err
+}
+
 // Schema returns information about each index in Pilosa including which fields
 // they contain.
 func (api *API) Schema(ctx context.Context) []*IndexInfo {

http/handler.go

@@ -1783,8 +1783,12 @@ func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Reques
 	}
 	err := h.api.ClusterMessage(r.Context(), r.Body)
 	if err != nil {
-		// TODO this was the previous behavior, but perhaps not everything is a bad request
-		http.Error(w, err.Error(), http.StatusBadRequest)
+		switch err := err.(type) {
+		case pilosa.MessageProcessingError:
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+		default:
+			http.Error(w, err.Error(), http.StatusBadRequest)
+		}
 		return
 	}