Check partition count before updating routing table

Author: buraksezer

cmd/olricd/server/server.go

@@ -153,7 +153,6 @@ func New(c *Config) (*Olricd, error) {
 
 	s.log = log.New(logOutput, "", log.LstdFlags)
 	s.config = &config.Config{
-		Name:                c.Olricd.Name, // deprecated, it will be removed in v0.3.0
 		BindAddr:            c.Olricd.BindAddr,
 		BindPort:            c.Olricd.BindPort,
 		Interface:           c.Olricd.Interface,

olric.go

@@ -55,6 +55,9 @@ var (
 
 	// ErrUnknownOperation means that an unidentified message has been received from a client.
 	ErrUnknownOperation = errors.New("unknown operation")
+
+	// ErrBadRequest denotes that request body is invalid.
+	ErrBadRequest = errors.New("bad request")
 )
 
 // ReleaseVersion is the current stable version of Olric
@@ -447,6 +450,8 @@ func (db *Olric) prepareResponse(req *protocol.Message, err error) *protocol.Mes
 		return req.Error(protocol.StatusErrUnknownOperation, err)
 	case err == ErrEndOfQuery:
 		return req.Error(protocol.StatusErrEndOfQuery, err)
+	case  err == ErrBadRequest:
+		return req.Error(protocol.StatusBadRequest, err)
 	default:
 		return req.Error(protocol.StatusInternalServerError, err)
 	}
@@ -483,6 +488,8 @@ func (db *Olric) requestTo(addr string, opcode protocol.OpCode, req *protocol.Me
 		return nil, ErrEndOfQuery
 	case resp.Status == protocol.StatusErrUnknownOperation:
 		return nil, ErrUnknownOperation
+	case resp.Status == protocol.StatusBadRequest:
+		return nil, ErrBadRequest
 	}
 	return nil, fmt.Errorf("unknown status code: %d", resp.Status)
 }

routing.go

@@ -444,7 +444,14 @@ func (db *Olric) updateRoutingOperation(req *protocol.Message) *protocol.Message
 	coordinator, err := db.checkAndGetCoordinator(coordinatorID)
 	if err != nil {
 		db.log.V(2).Printf("[ERROR] Routing table cannot be updated: %v", err)
-		return req.Error(protocol.StatusInternalServerError, err)
+		return req.Error(protocol.StatusBadRequest, err)
+	}
+
+	// Compare partition counts to catch a possible inconsistencies in configuration
+	if db.config.PartitionCount != uint64(len(table)) {
+		db.log.V(2).Printf("[ERROR] Routing table cannot be updated. " +
+			"Expected partition count is %d, got: %d", db.config.PartitionCount, uint64(len(table)))
+		return req.Error(protocol.StatusBadRequest, err)
 	}
 
 	// owners(atomic.Value) is guarded by routingUpdateMtx against parallel writers.