Repository: weaviate/weaviate
Pull request #2674 (2023-02-23 17:33:09)

Additional tunable consistency support for objects PUT, HEAD
### What's being changed:

- Fix objects PUT tunable consistency bug
  - only default ALL consistency level was being used
- Implement HEAD (exists) tunable consistency
  - fix replication coordinator read exists bug

### Review checklist

- [ ] Documentation has been updated, if necessary. Link to changed documentation:
- [ ] Chaos pipeline run or not necessary. Link to pipeline:
- [ ] All new code is covered by tests where it is reasonable.
- [ ] Performance tests have been run or not necessary.



Patch:
diff --git a/adapters/handlers/rest/clusterapi/indices_replicas.go b/adapters/handlers/rest/clusterapi/indices_replicas.go
index 9dbeaa718b..3487cf8f6b 100644
--- a/adapters/handlers/rest/clusterapi/indices_replicas.go
+++ b/adapters/handlers/rest/clusterapi/indices_replicas.go
@@ -50,8 +50,6 @@ type replicator interface {
 	// Read endpoints
 	FetchObject(ctx context.Context, indexName,
 		shardName string, id strfmt.UUID) (objects.Replica, error)
-	DoesExist(ctx context.Context, class,
-		shardName string, id strfmt.UUID) (objects.Replica, error)
 	FetchObjects(ctx context.Context, class,
 		shardName string, ids []strfmt.UUID) ([]objects.Replica, error)
 	DigestObjects(ctx context.Context, class, shardName string,
@@ -574,18 +572,10 @@ func (i *replicatedIndices) getObject() http.Handler {
 			err  error
 		)
 
-		if r.URL.Query().Get("check_exists") != "" {
-			resp, err = i.shards.DoesExist(r.Context(), index, shard, strfmt.UUID(id))
-			if err != nil {
-				http.Error(w, err.Error(), http.StatusInternalServerError)
-				return
-			}
-		} else {
-			resp, err = i.shards.FetchObject(r.Context(), index, shard, strfmt.UUID(id))
-			if err != nil {
-				http.Error(w, err.Error(), http.StatusInternalServerError)
-				return
-			}
+		resp, err = i.shards.FetchObject(r.Context(), index, shard, strfmt.UUID(id))
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
 		}
 
 		b, err := resp.MarshalBinary()
diff --git a/adapters/handlers/rest/embedded_spec.go b/adapters/handlers/rest/embedded_spec.go
index 22e5846b49..fe0628cf61 100644
--- a/adapters/handlers/rest/embedded_spec.go
+++ b/adapters/handlers/rest/embedded_spec.go
@@ -1330,6 +1330,9 @@ func init() {
             "name": "id",
             "in": "path",
             "required": true
+          },
+          {
+            "$ref": "#/parameters/CommonConsistencyLevelParameterQuery"
           }
         ],
         "responses": {
@@ -5648,6 +5651,12 @@ func init() {
             "name": "id",
             "in": "path",
             "required": true
+          },
+          {
+            "type": "string",
+            "description": "Determines how many replicas must acknowledge a request before it is considered successful",
+            "name": "consistency_level",
+            "in": "query"
           }
         ],
         "responses": {
diff --git a/adapters/handlers/rest/handlers_objects.go b/adapters/handlers/rest/handlers_objects.go
index 250e2be77e..b175cbfd80 100644
--- a/adapters/handlers/rest/handlers_objects.go
+++ b/adapters/handlers/rest/handlers_objects.go
@@ -46,14 +46,15 @@ type ModulesProvider interface {
 type objectsManager interface {
 	AddObject(context.Context, *models.Principal, *models.Object,
 		*additional.ReplicationProperties) (*models.Object, error)
-	ValidateObject(context.Context, *models.Principal, *models.Object) error
+	ValidateObject(context.Context, *models.Principal, *models.Object, *additional.ReplicationProperties) error
 	GetObject(_ context.Context, _ *models.Principal, class string, _ strfmt.UUID,
 		_ additional.Properties, _ *additional.ReplicationProperties) (*models.Object, error)
 	DeleteObject(_ context.Context, _ *models.Principal,
 		class string, _ strfmt.UUID, _ *additional.ReplicationProperties) error
 	UpdateObject(_ context.Context, _ *models.Principal, class string, _ strfmt.UUID,
 		_ *models.Object, _ *additional.ReplicationProperties) (*models.Object, error)
-	HeadObject(_ context.Context, _ *models.Principal, class string, _ strfmt.UUID) (bool, *uco.Error)
+	HeadObject(ctx context.Context, principal *models.Principal, class string,
+		id strfmt.UUID, repl *additional.ReplicationProperties) (bool, *uco.Error)
 	GetObjects(context.Context, *models.Principal, *int64, *int64, *string, *string, additional.Properties) ([]*models.Object, error)
 	Query(ctx context.Context, principal *models.Principal, params *uco.QueryParams) ([]*models.Object, *uco.Error)
 	MergeObject(context.Context, *models.Principal, *models.Object, *additional.ReplicationProperties) *uco.Error
@@ -100,7 +101,7 @@ func (h *objectHandlers) addObject(params objects.ObjectsCreateParams,
 func (h *objectHandlers) validateObject(params objects.ObjectsValidateParams,
 	principal *models.Principal,
 ) middleware.Responder {
-	err := h.manager.ValidateObject(params.HTTPRequest.Context(), principal, params.Body)
+	err := h.manager.ValidateObject(params.HTTPRequest.Context(), principal, params.Body, nil)
 	if err != nil {
 		switch err.(type) {
 		case errors.Forbidden:
@@ -327,22 +328,29 @@ func (h *objectHandlers) updateObject(params objects.ObjectsClassPutParams,
 	return objects.NewObjectsClassPutOK().WithPayload(object)
 }
 
-func (h *objectHandlers) headObject(r objects.ObjectsClassHeadParams,
+func (h *objectHandlers) headObject(params objects.ObjectsClassHeadParams,
 	principal *models.Principal,
 ) middleware.Responder {
-	ok, err := h.manager.HeadObject(r.HTTPRequest.Context(), principal, r.ClassName, r.ID)
+	repl, err := getReplicationProperties(params.ConsistencyLevel, nil)
 	if err != nil {
+		return objects.NewObjectsCreateBadRequest().
+			WithPayload(errPayloadFromSingleErr(err))
+	}
+
+	exists, objErr := h.manager.HeadObject(params.HTTPRequest.Context(),
+		principal, params.ClassName, params.ID, repl)
+	if objErr != nil {
 		switch {
-		case err.Forbidden():
+		case objErr.Forbidden():
 			return objects.NewObjectsClassHeadForbidden().
-				WithPayload(errPayloadFromSingleErr(err))
+				WithPayload(errPayloadFromSingleErr(objErr))
 		default:
 			return objects.NewObjectsClassHeadInternalServerError().
-				WithPayload(errPayloadFromSingleErr(err))
+				WithPayload(errPayloadFromSingleErr(objErr))
 		}
 	}
 
-	if !ok {
+	if !exists {
 		return objects.NewObjectsClassHeadNotFound()
 	}
 	return objects.NewObjectsClassHeadNoContent()
diff --git a/adapters/handlers/rest/operations/objects/objects_class_head_parameters.go b/adapters/handlers/rest/operations/objects/objects_class_head_parameters.go
index 4c349a6681..5ec564bf6e 100644
--- a/adapters/handlers/rest/operations/objects/objects_class_head_parameters.go
+++ b/adapters/handlers/rest/operations/objects/objects_class_head_parameters.go
@@ -20,6 +20,7 @@ import (
 	"net/http"
 
 	"github.com/go-openapi/errors"
+	"github.com/go-openapi/runtime"
 	"github.com/go-openapi/runtime/middleware"
 	"github.com/go-openapi/strfmt"
 	"github.com/go-openapi/validate"
@@ -47,6 +48,10 @@ type ObjectsClassHeadParams struct {
 	  In: path
 	*/
 	ClassName string
+	/*Determines how many replicas must acknowledge a request before it is considered successful
+	  In: query
+	*/
+	ConsistencyLevel *string
 	/*The uuid of the data object
 	  Required: true
 	  In: path
@@ -63,11 +68,18 @@ func (o *ObjectsClassHeadParams) BindRequest(r *http.Request, route *middleware.
 
 	o.HTTPRequest = r
 
+	qs := runtime.Values(r.URL.Query())
+
 	rClassName, rhkClassName, _ := route.Params.GetOK("className")
 	if err := o.bindClassName(rClassName, rhkClassName, route.Formats); err != nil {
 		res = append(res, err)
 	}
 
+	qConsistencyLevel, qhkConsistencyLevel, _ := qs.GetOK("consistency_level")
+	if err := o.bindConsistencyLevel(qConsistencyLevel, qhkConsistencyLevel, route.Formats); err != nil {
+		res = append(res, err)
+	}
+
 	rID, rhkID, _ := route.Params.GetOK("id")
 	if err := o.bindID(rID, rhkID, route.Formats); err != nil {
 		res = append(res, err)
@@ -92,6 +104,24 @@ func (o *ObjectsClassHeadParams) bindClassName(rawData []string, hasKey bool, fo
 	return nil
 }
 
+// bindConsistencyLevel binds and validates parameter ConsistencyLevel from query.
+func (o *ObjectsClassHeadParams) bindConsistencyLevel(rawData []string, hasKey bool, formats strfmt.Registry) error {
+	var raw string
+	if len(rawData) > 0 {
+		raw = rawData[len(rawData)-1]
+	}
+
+	// Required: false
+	// AllowEmptyValue: false
+
+	if raw == "" { // empty values pass all other validations
+		return nil
+	}
+	o.ConsistencyLevel = &raw
+
+	return nil
+}
+
 // bindID binds and validates parameter ID from path.
 func (o *ObjectsClassHeadParams) bindID(rawData []string, hasKey bool, formats strfmt.Registry) error {
 	var raw string
diff --git a/adapters/handlers/rest/operations/objects/objects_class_head_urlbuilder.go b/adapters/handlers/rest/operations/objects/objects_class_head_urlbuilder.go
index 20823cdbd2..b916ef4933 100644
--- a/adapters/handlers/rest/operations/objects/objects_class_head_urlbuilder.go
+++ b/adapters/handlers/rest/operations/objects/objects_class_head_urlbuilder.go
@@ -30,6 +30,8 @@ type ObjectsClassHeadURL struct {
 	ClassName string
 	ID        strfmt.UUID
 
+	ConsistencyLevel *string
+
 	_basePath string
 	// avoid unkeyed usage
 	_ struct{}
@@ -76,6 +78,18 @@ func (o *ObjectsClassHeadURL) Build() (*url.URL, error) {
 	}
 	_result.Path = golangswaggerpaths.Join(_basePath, _path)
 
+	qs := make(url.Values)
+
+	var consistencyLevelQ string
+	if o.ConsistencyLevel != nil {
+		consistencyLevelQ = *o.ConsistencyLevel
+	}
+	if consistencyLevelQ != "" {
+		qs.Set("consistency_level", consistencyLevelQ)
+	}
+
+	_result.RawQuery = qs.Encode()
+
 	return &_result, nil
 }
 
diff --git a/adapters/repos/db/crud.go b/adapters/repos/db/crud.go
index 59ff9ac5d1..0d2585a94d 100644
--- a/adapters/repos/db/crud.go
+++ b/adapters/repos/db/crud.go
@@ -190,25 +190,29 @@ func (d *DB) enrichRefsForSingle(ctx context.Context, obj *search.Result,
 	return &res[0], nil
 }
 
-func (d *DB) Exists(ctx context.Context, class string, id strfmt.UUID) (bool, error) {
+func (d *DB) Exists(ctx context.Context, class string,
+	id strfmt.UUID, repl *additional.ReplicationProperties,
+) (bool, error) {
 	if class == "" {
-		return d.anyExists(ctx, id)
+		return d.anyExists(ctx, id, repl)
 	}
 	index := d.GetIndex(schema.ClassName(class))
 	if index == nil {
 		return false, nil
 	}
-	return index.exists(ctx, id)
+	return index.exists(ctx, id, repl)
 }
 
-func (d *DB) anyExists(ctx context.Context, id strfmt.UUID) (bool, error) {
+func (d *DB) anyExists(ctx context.Context, id strfmt.UUID,
+	repl *additional.ReplicationProperties,
+) (bool, error) {
 	// TODO: Search in parallel, rather than sequentially or this will be
 	// painfully slow on large schemas
 	d.indexLock.RLock()
 	defer d.indexLock.RUnlock()
 
 	for _, index := range d.indices {
-		ok, err := index.exists(ctx, id)
+		ok, err := index.exists(ctx, id, repl)
 		if err != nil {
 			return false, errors.Wrapf(err, "search index %s", index.ID())
 		}
diff --git a/adapters/repos/db/index.go b/adapters/repos/db/index.go
index 3a637c63d5..62452d77dd 100644
--- a/adapters/repos/db/index.go
+++ b/adapters/repos/db/index.go
@@ -735,28 +735,35 @@ func wrapIDsInMulti(in []strfmt.UUID) []multi.Identifier {
 	return out
 }
 
-func (i *Index) exists(ctx context.Context, id strfmt.UUID) (bool, error) {
+func (i *Index) exists(ctx context.Context, id strfmt.UUID,
+	replProps *additional.ReplicationProperties,
+) (bool, error) {
 	shardName, err := i.shardFromUUID(id)
 	if err != nil {
 		return false, err
 	}
 
-	local := i.getSchema.
-		ShardingState(i.Config.ClassName.String()).
-		IsShardLocal(shardName)
-
-	var ok bool
-	if local {
+	var exists bool
+	if i.replicationEnabled() {
+		if replProps == nil {
+			replProps = &additional.ReplicationProperties{
+				ConsistencyLevel: string(replica.All),
+			}
+		}
+		exists, err = i.replicator.Exists(ctx,
+			replica.ConsistencyLevel(replProps.ConsistencyLevel), shardName, id)
+	} else if i.isLocalShard(shardName) {
 		shard := i.Shards[shardName]
-		ok, err = shard.exists(ctx, id)
+		exists, err = shard.exists(ctx, id)
 	} else {
-		ok, err = i.remote.Exists(ctx, shardName, id)
+		exists, err = i.remote.Exists(ctx, shardName, id)
 	}
+
 	if err != nil {
 		return false, errors.Wrapf(err, "shard %s", shardName)
 	}
 
-	return ok, nil
+	return exists, nil
 }
 
 func (i *Index) IncomingExists(ctx context.Context, shardName string,
diff --git a/adapters/repos/db/replication.go b/adapters/repos/db/replication.go
index 1e87e69f8d..869c8a4f90 100644
--- a/adapters/repos/db/replication.go
+++ b/adapters/repos/db/replication.go
@@ -525,9 +525,3 @@ func (i *Index) fetchObjects(ctx context.Context,
 
 	return resp, nil
 }
-
-func (db *DB) DoesExist(ctx context.Context,
-	class, shardName string, id strfmt.UUID,
-) (objects.Replica, error) {
-	panic("not implemented")
-}
diff --git a/client/objects/objects_class_head_parameters.go b/client/objects/objects_class_head_parameters.go
index 852c7a84e9..666abfae8b 100644
--- a/client/objects/objects_class_head_parameters.go
+++ b/client/objects/objects_class_head_parameters.go
@@ -78,6 +78,12 @@ type ObjectsClassHeadParams struct {
 	*/
 	ClassName string
 
+	/* ConsistencyLevel.
+
+	   Determines how many replicas must acknowledge a request before it is considered successful
+	*/
+	ConsistencyLevel *string
+
 	/* ID.
 
 	   The uuid of the data object
@@ -150,6 +156,17 @@ func (o *ObjectsClassHeadParams) SetClassName(className string) {
 	o.ClassName = className
 }
 
+// WithConsistencyLevel adds the consistencyLevel to the objects class head params
+func (o *ObjectsClassHeadParams) WithConsistencyLevel(consistencyLevel *string) *ObjectsClassHeadParams {
+	o.SetConsistencyLevel(consistencyLevel)
+	return o
+}
+
+// SetConsistencyLevel adds the consistencyLevel to the objects class head params
+func (o *ObjectsClassHeadParams) SetConsistencyLevel(consistencyLevel *string) {
+	o.ConsistencyLevel = consistencyLevel
+}
+
 // WithID adds the id to the objects class head params
 func (o *ObjectsClassHeadParams) WithID(id strfmt.UUID) *ObjectsClassHeadParams {
 	o.SetID(id)
@@ -174,6 +191,23 @@ func (o *ObjectsClassHeadParams) WriteToRequest(r runtime.ClientRequest, reg str
 		return err
 	}
 
+	if o.ConsistencyLevel != nil {
+
+		// query param consistency_level
+		var qrConsistencyLevel string
+
+		if o.ConsistencyLevel != nil {
+			qrConsistencyLevel = *o.ConsistencyLevel
+		}
+		qConsistencyLevel := qrConsistencyLevel
+		if qConsistencyLevel != "" {
+
+			if err := r.SetQueryParam("consistency_level", qConsistencyLevel); err != nil {
+				return err
+			}
+		}
+	}
+
 	// path param id
 	if err := r.SetPathParam("id", o.ID.String()); err != nil {
 		return err
diff --git a/openapi-specs/schema.json b/openapi-specs/schema.json
index 0bba0b19f2..a7b8059adb 100644
--- a/openapi-specs/schema.json
+++ b/openapi-specs/schema.json
@@ -2507,6 +2507,9 @@
             "name": "id",
             "required": true,
             "type": "string"
+          },
+          {
+            "$ref": "#/parameters/CommonConsistencyLevelParameterQuery"
           }
         ],
         "responses": {
diff --git a/usecases/objects/add.go b/usecases/objects/add.go
index 952c69eea2..dc0a975b2c 100644
--- a/usecases/objects/add.go
+++ b/usecases/objects/add.go
@@ -15,10 +15,9 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/weaviate/weaviate/entities/additional"
-
 	"github.com/go-openapi/strfmt"
 	"github.com/google/uuid"
+	"github.com/weaviate/weaviate/entities/additional"
 	"github.com/weaviate/weaviate/entities/models"
 	"github.com/weaviate/weaviate/entities/schema"
 	"github.com/weaviate/weaviate/usecases/objects/validation"
@@ -57,7 +56,7 @@ func (m *Manager) AddObject(ctx context.Context, principal *models.Principal,
 }
 
 func (m *Manager) checkIDOrAssignNew(ctx context.Context, class string,
-	id strfmt.UUID,
+	id strfmt.UUID, repl *additional.ReplicationProperties,
 ) (strfmt.UUID, error) {
 	if id == "" {
 		newID, err := generateUUID()
@@ -68,7 +67,7 @@ func (m *Manager) checkIDOrAssignNew(ctx context.Context, class string,
 	}
 
 	// only validate ID uniqueness if explicitly set
-	if ok, err := m.vectorRepo.Exists(ctx, class, id); ok {
+	if ok, err := m.vectorRepo.Exists(ctx, class, id, repl); ok {
 		return "", NewErrInvalidUserInput("id '%s' already exists", id)
 	} else if err != nil {
 		return "", NewErrInternal(err.Error())
@@ -79,7 +78,7 @@ func (m *Manager) checkIDOrAssignNew(ctx context.Context, class string,
 func (m *Manager) addObjectToConnectorAndSchema(ctx context.Context, principal *models.Principal,
 	object *models.Object, repl *additional.ReplicationProperties,
 ) (*models.Object, error) {
-	id, err := m.checkIDOrAssignNew(ctx, object.Class, object.ID)
+	id, err := m.checkIDOrAssignNew(ctx, object.Class, object.ID, repl)
 	if err != nil {
 		return nil, err
 	}
@@ -90,7 +89,7 @@ func (m *Manager) addObjectToConnectorAndSchema(ctx context.Context, principal *
 		return nil, NewErrInvalidUserInput("invalid object: %v", err)
 	}
 
-	err = m.validateObject(ctx, principal, object)
+	err = m.validateObject(ctx, principal, object, repl)
 	if err != nil {
 		return nil, NewErrInvalidUserInput("invalid object: %v", err)
 	}
@@ -118,7 +117,10 @@ func (m *Manager) addObjectToConnectorAndSchema(ctx context.Context, principal *
 	return object, nil
 }
 
-func (m *Manager) validateObject(ctx context.Context, principal *models.Principal, object *models.Object) error {
+func (m *Manager) validateObject(ctx context.Context,
+	principal *models.Principal, object *models.Object,
+	repl *additional.ReplicationProperties,
+) error {
 	// Validate schema given in body with the weaviate schema
 	if _, err := uuid.Parse(object.ID.String()); err != nil {
 		return err
@@ -129,5 +131,5 @@ func (m *Manager) validateObject(ctx context.Context, principal *models.Principa
 		return err
 	}
 
-	return validation.New(m.vectorRepo.Exists, m.config).Object(ctx, object, class)
+	return validation.New(m.vectorRepo.Exists, m.config, repl).Object(ctx, object, class)
 }
diff --git a/usecases/objects/batch_add.go b/usecases/objects/batch_add.go
index 738332cc3e..ccfc576c5e 100644
--- a/usecases/objects/batch_add.go
+++ b/usecases/objects/batch_add.go
@@ -56,7 +56,7 @@ func (b *BatchManager) addObjects(ctx context.Context, principal *models.Princip
 		return nil, NewErrInvalidUserInput("invalid param 'objects': %v", err)
 	}
 
-	batchObjects := b.validateObjectsConcurrently(ctx, principal, classes, fields)
+	batchObjects := b.validateObjectsConcurrently(ctx, principal, classes, fields, repl)
 	b.metrics.BatchOp("total_preprocessing", beforePreProcessing.UnixNano())
 
 	var (
@@ -82,7 +82,7 @@ func (b *BatchManager) validateObjectForm(classes []*models.Object) error {
 }
 
 func (b *BatchManager) validateObjectsConcurrently(ctx context.Context, principal *models.Principal,
-	classes []*models.Object, fields []*string,
+	classes []*models.Object, fields []*string, repl *additional.ReplicationProperties,
 ) BatchObjects {
 	fieldsToKeep := determineResponseFields(fields)
 	c := make(chan BatchObject, len(classes))
@@ -92,7 +92,7 @@ func (b *BatchManager) validateObjectsConcurrently(ctx context.Context, principa
 	// Generate a goroutine for each separate request
 	for i, object := range classes {
 		wg.Add(1)
-		go b.validateObject(ctx, principal, wg, object, i, &c, fieldsToKeep)
+		go b.validateObject(ctx, principal, wg, object, i, &c, fieldsToKeep, repl)
 	}
 
 	wg.Wait()
@@ -102,7 +102,7 @@ func (b *BatchManager) validateObjectsConcurrently(ctx context.Context, principa
 
 func (b *BatchManager) validateObject(ctx context.Context, principal *models.Principal,
 	wg *sync.WaitGroup, concept *models.Object, originalIndex int, resultsC *chan BatchObject,
-	fieldsToKeep map[string]struct{},
+	fieldsToKeep map[string]struct{}, repl *additional.ReplicationProperties,
 ) {
 	defer wg.Done()
 
@@ -155,7 +155,7 @@ func (b *BatchManager) validateObject(ctx context.Context, principal *models.Pri
 		ec.Add(fmt.Errorf("class '%s' not present in schema", object.Class))
 	} else {
 		// not possible without the class being present
-		err = validation.New(b.vectorRepo.Exists, b.config).Object(ctx, object, class)
+		err = validation.New(b.vectorRepo.Exists, b.config, repl).Object(ctx, object, class)
 		ec.Add(err)
 
 		err = b.modulesProvider.UpdateVector(ctx, object, class, nil, b.findObject, b.logger)
diff --git a/usecases/objects/delete.go b/usecases/objects/delete.go
index 98e85a9829..d3ab933f10 100644
--- a/usecases/objects/delete.go
+++ b/usecases/objects/delete.go
@@ -49,7 +49,7 @@ func (m *Manager) DeleteObject(ctx context.Context, principal *models.Principal,
 		return m.deleteObjectFromRepo(ctx, id)
 	}
 
-	ok, err := m.vectorRepo.Exists(ctx, class, id)
+	ok, err := m.vectorRepo.Exists(ctx, class, id, repl)
 	if err != nil {
 		return NewErrInternal("check object existence: %v", err)
 	}
diff --git a/usecases/objects/head.go b/usecases/objects/head.go
index 16d528d14a..4957d9cbc6 100644
--- a/usecases/objects/head.go
+++ b/usecases/objects/head.go
@@ -16,11 +16,15 @@ import (
 	"fmt"
 
 	"github.com/go-openapi/strfmt"
+	"github.com/weaviate/weaviate/entities/additional"
 	"github.com/weaviate/weaviate/entities/models"
 )
 
 // HeadObject check object's existence in the conncected DB
-func (m *Manager) HeadObject(ctx context.Context, principal *models.Principal, class string, id strfmt.UUID) (bool, *Error) {
+func (m *Manager) HeadObject(ctx context.Context,
+	principal *models.Principal, class string, id strfmt.UUID,
+	repl *additional.ReplicationProperties,
+) (bool, *Error) {
 	path := fmt.Sprintf("objects/%s", id)
 	if class != "" {
 		path = fmt.Sprintf("objects/%s/%s", class, id)
@@ -38,7 +42,7 @@ func (m *Manager) HeadObject(ctx context.Context, principal *models.Principal, c
 	m.metrics.HeadObjectInc()
 	defer m.metrics.HeadObjectDec()
 
-	ok, err := m.vectorRepo.Exists(ctx, class, id)
+	ok, err := m.vectorRepo.Exists(ctx, class, id, repl)
 	if err != nil {
 		return false, &Error{"repo.exists", StatusInternalServerError, err}
 	}
diff --git a/usecases/objects/manager.go b/usecases/objects/manager.go
index acea5100f9..d071d2a724 100644
--- a/usecases/objects/manager.go
+++ b/usecases/objects/manager.go
@@ -96,7 +96,7 @@ type VectorRepo interface {
 	Object(ctx context.Context, class string, id strfmt.UUID, props search.SelectProperties,
 		additional additional.Properties, repl *additional.ReplicationProperties) (*search.Result, error)
 	// Exists returns true if an object of a giving class exists
-	Exists(ctx context.Context, class string, id strfmt.UUID) (bool, error)
+	Exists(ctx context.Context, class string, id strfmt.UUID, repl *additional.ReplicationProperties) (bool, error)
 	ObjectByID(ctx context.Context, id strfmt.UUID, props search.SelectProperties,
 		additional additional.Properties) (*search.Result, error)
 	ObjectSearch(ctx context.Context, offset, limit int, filters *filters.LocalFilter,
diff --git a/usecases/objects/merge.go b/usecases/objects/merge.go
index 63e48f3737..614f426d0c 100644
--- a/usecases/objects/merge.go
+++ b/usecases/objects/merge.go
@@ -50,7 +50,7 @@ func (m *Manager) MergeObject(ctx context.Context, principal *models.Principal,
 	m.metrics.MergeObjectInc()
 	defer m.metrics.MergeObjectDec()
 
-	if err := m.validateObject(ctx, principal, updates); err != nil {
+	if err := m.validateObject(ctx, principal, updates, repl); err != nil {
 		return &Error{"bad request", StatusBadRequest, err}
 	}
 	if updates.Properties == nil {
diff --git a/usecases/objects/references_add.go b/usecases/objects/references_add.go
index cecaf90d04..2b6ed1888d 100644
--- a/usecases/objects/references_add.go
+++ b/usecases/objects/references_add.go
@@ -57,12 +57,12 @@ func (m *Manager) AddObjectReference(
 	}
 	defer unlock()
 
-	validator := validation.New(m.vectorRepo.Exists, m.config)
+	validator := validation.New(m.vectorRepo.Exists, m.config, repl)
 	if err := input.validate(ctx, principal, validator, m.schemaManager); err != nil {
 		return &Error{"validate inputs", StatusBadRequest, err}
 	}
 	if !deprecatedEndpoint {
-		ok, err := m.vectorRepo.Exists(ctx, input.Class, input.ID)
+		ok, err := m.vectorRepo.Exists(ctx, input.Class, input.ID, repl)
 		if err != nil {
 			return &Error{"source object", StatusInternalServerError, err}
 		}
diff --git a/usecases/objects/references_update.go b/usecases/objects/references_update.go
index cbf504d337..906575b040 100644
--- a/usecases/objects/references_update.go
+++ b/usecases/objects/references_update.go
@@ -71,7 +71,7 @@ func (m *Manager) UpdateObjectReferences(
 	}
 	defer unlock()
 
-	validator := validation.New(m.vectorRepo.Exists, m.config)
+	validator := validation.New(m.vectorRepo.Exists, m.config, repl)
 	if err := input.validate(ctx, principal, validator, m.schemaManager); err != nil {
 		return &Error{"bad inputs", StatusBadRequest, err}
 	}
diff --git a/usecases/objects/update.go b/usecases/objects/update.go
index fd20ca995b..a26bb7c7e5 100644
--- a/usecases/objects/update.go
+++ b/usecases/objects/update.go
@@ -55,7 +55,7 @@ func (m *Manager) updateObjectToConnectorAndSchema(ctx context.Context, principa
 		return nil, NewErrInvalidUserInput("invalid update: field 'id' is immutable")
 	}
 
-	obj, err := m.getObjectFromRepo(ctx, className, id, additional.Properties{}, nil)
+	obj, err := m.getObjectFromRepo(ctx, className, id, additional.Properties{}, repl)
 	if err != nil {
 		return nil, err
 	}
@@ -67,7 +67,7 @@ func (m *Manager) updateObjectToConnectorAndSchema(ctx context.Context, principa
 		WithField("id", id).
 		Debug("received update kind request")
 
-	err = m.validateObject(ctx, principal, updates)
+	err = m.validateObject(ctx, principal, updates, repl)
 	if err != nil {
 		return nil, NewErrInvalidUserInput("invalid object: %v", err)
 	}
diff --git a/usecases/objects/validate.go b/usecases/objects/validate.go
index e846ac5fbb..c17d265837 100644
--- a/usecases/objects/validate.go
+++ b/usecases/objects/validate.go
@@ -14,13 +14,14 @@ package objects
 import (
 	"context"
 
+	"github.com/weaviate/weaviate/entities/additional"
 	"github.com/weaviate/weaviate/entities/models"
 )
 
 // ValidateObject without adding it to the database. Can be used in UIs for
 // async validation before submitting
 func (m *Manager) ValidateObject(ctx context.Context, principal *models.Principal,
-	class *models.Object,
+	class *models.Object, repl *additional.ReplicationProperties,
 ) error {
 	err := m.authorizer.Authorize(principal, "validate", "objects")
 	if err != nil {
@@ -33,7 +34,7 @@ func (m *Manager) ValidateObject(ctx context.Context, principal *models.Principa
 	}
 	defer unlock()
 
-	err = m.validateObject(ctx, principal, class)
+	err = m.validateObject(ctx, principal, class, repl)
 	if err != nil {
 		return NewErrInvalidUserInput("invalid object: %v", err)
 	}
diff --git a/usecases/objects/validation/model_validation.go b/usecases/objects/validation/model_validation.go
index 11feab3057..e1f1215c6b 100644
--- a/usecases/objects/validation/model_validation.go
+++ b/usecases/objects/validation/model_validation.go
@@ -16,12 +16,13 @@ import (
 	"fmt"
 
 	"github.com/go-openapi/strfmt"
+	"github.com/weaviate/weaviate/entities/additional"
 	"github.com/weaviate/weaviate/entities/models"
 	"github.com/weaviate/weaviate/entities/schema/crossref"
 	"github.com/weaviate/weaviate/usecases/config"
 )
 
-type exists func(_ context.Context, class string, _ strfmt.UUID) (bool, error)
+type exists func(_ context.Context, class string, _ strfmt.UUID, _ *additional.ReplicationProperties) (bool, error)
 
 const (
 	// ErrorMissingActionObjects message
@@ -59,16 +60,18 @@ const (
 )
 
 type Validator struct {
-	exists exists
-	config *config.WeaviateConfig
+	exists           exists
+	config           *config.WeaviateConfig
+	replicationProps *additional.ReplicationProperties
 }
 
-func New(exists exists,
-	config *config.WeaviateConfig,
+func New(exists exists, config *config.WeaviateConfig,
+	repl *additional.ReplicationProperties,
 ) *Validator {
 	return &Validator{
-		exists: exists,
-		config: config,
+		exists:           exists,
+		config:           config,
+		replicationProps: repl,
 	}
 }
 
@@ -104,7 +107,7 @@ func (v *Validator) ValidateSingleRef(ctx context.Context, cref *models.SingleRe
 	}
 
 	// locally check for object existence
-	ok, err := v.exists(ctx, ref.Class, ref.TargetID)
+	ok, err := v.exists(ctx, ref.Class, ref.TargetID, v.replicationProps)
 	if err != nil {
 		return err
 	}
diff --git a/usecases/replica/finder.go b/usecases/replica/finder.go
index ee46794de2..e2820084c6 100644
--- a/usecases/replica/finder.go
+++ b/usecases/replica/finder.go
@@ -23,6 +23,7 @@ import (
 	"github.com/weaviate/weaviate/entities/search"
 	"github.com/weaviate/weaviate/entities/storobj"
 	"github.com/weaviate/weaviate/usecases/objects"
+	"golang.org/x/sync/errgroup"
 )
 
 var (
@@ -170,7 +171,7 @@ func (f *Finder) Exists(ctx context.Context, l ConsistencyLevel, shard string, i
 			x = xs[0]
 		}
 		if err == nil && len(xs) != 1 {
-			err = fmt.Errorf("digest read request: empty result")
+			err = fmt.Errorf("malformed digest read response: expected one result, got %d", len(xs))
 		}
 		return existReply{host, x}, err
 	}
@@ -286,23 +287,29 @@ func (f *Finder) repairOne(ctx context.Context, shard string, id strfmt.UUID, vo
 		}
 	}
 
-	for _, c := range votes { // repair
-		if c.UTime != lastUTime {
-			updates := []*objects.VObject{{
+	var gr errgroup.Group
+	for _, vote := range votes { // repair
+		if vote.UTime == lastUTime {
+			continue
+		}
+		vote := vote
+		gr.Go(func() error {
+			ups := []*objects.VObject{{
 				LatestObject:    &updates.Object.Object,
-				StaleUpdateTime: c.UTime,
-				Version:         0, // todo set when implemented
+				StaleUpdateTime: vote.UTime,
 			}}
-			resp, err := f.RClient.OverwriteObjects(ctx, c.sender, f.class, shard, updates)
+			resp, err := f.RClient.OverwriteObjects(ctx, vote.sender, f.class, shard, ups)
 			if err != nil {
-				return nil, fmt.Errorf("node %q could not repair object: %w", c.sender, err)
+				return fmt.Errorf("node %q could not repair object: %w", vote.sender, err)
 			}
 			if len(resp) > 0 && resp[0].Err != "" && resp[0].UpdateTime != lastUTime {
-				return nil, fmt.Errorf("overwrite %w %s: %s", errConflictObjectChanged, c.sender, resp[0].Err)
+				return fmt.Errorf("overwrite %w %s: %s", errConflictObjectChanged, vote.sender, resp[0].Err)
 			}
-		}
+			return nil
+		})
 	}
-	return updates.Object, nil
+
+	return updates.Object, gr.Wait()
 }
 
 type vote struct {
@@ -367,6 +374,9 @@ func (f *Finder) readAll(ctx context.Context, shard string, ids []strfmt.UUID, c
 			}
 		}
 		res, err := f.repairAll(ctx, shard, ids, votes, st, contentIdx)
+		if err != nil {
+			res = nil
+		}
 		resultCh <- _Results{res, err}
 	}()
 
@@ -424,7 +434,9 @@ func (f *Finder) repairAll(ctx context.Context,
 			}
 		}
 		partitions = append(partitions, len(ms))
-		// TODO parallel fetch
+
+		// concurrent fetches
+		gr, ctx := errgroup.WithContext(ctx)
 		start := 0
 		for _, end := range partitions { // fetch diffs
 			receiver := votes[ms[start].S].Sender
@@ -433,28 +445,34 @@ func (f *Finder) repairAll(ctx context.Context,
 				query[j] = ids[ms[start].O]
 				j++
 			}
-			resp, err := f.RClient.FetchObjects(ctx, receiver, f.class, shard, query)
-			if err != nil {
-				return nil, err
-			}
-			n := len(query)
-			if m := len(resp); n != m {
-				return nil, fmt.Errorf("try to fetch %d objects from %s but got %d", m, receiver, n)
-			}
-			for i := 0; i < n; i++ {
-				idx := ms[start-n+i].O
-				if lastTimes[idx].T != resp[i].UpdateTime() {
-					return nil, fmt.Errorf("object %s changed on %s", ids[idx], receiver)
+			start := start
+			gr.Go(func() error {
+				resp, err := f.RClient.FetchObjects(ctx, receiver, f.class, shard, query)
+				if err != nil {
+					return err
+				}
+				n := len(query)
+				if m := len(resp); n != m {
+					return fmt.Errorf("try to fetch %d objects from %s but got %d", m, receiver, n)
+				}
+				for i := 0; i < n; i++ {
+					idx := ms[start-n+i].O
+					if lastTimes[idx].T != resp[i].UpdateTime() {
+						return fmt.Errorf("object %s changed on %s", ids[idx], receiver)
+					}
+					result[idx] = resp[i].Object
 				}
-				result[idx] = resp[i].Object
+				return nil
+			})
+			if err := gr.Wait(); err != nil {
+				return nil, err
 			}
 		}
 	}
 
-	// repair
-	// TODO parallel overwrite
+	// concurrent repairs
+	gr, ctx := errgroup.WithContext(ctx)
 	for _, vote := range votes {
-		receiver := vote.Sender
 		query := make([]*objects.VObject, 0, len(ids)/2)
 		for j, x := range lastTimes {
 			if cTime := vote.UpdateTimeAt(j); x.T != cTime && !x.Deleted {
@@ -464,21 +482,26 @@ func (f *Finder) repairAll(ctx context.Context,
 		if len(query) == 0 {
 			continue
 		}
-		rs, err := f.RClient.OverwriteObjects(ctx, receiver, f.class, shard, query)
-		if err != nil {
-			return nil, fmt.Errorf("node %q could not repair objects: %w", receiver, err)
-		}
-		for _, r := range rs {
-			if r.Err != "" {
-				return nil, fmt.Errorf("object changed in the meantime on node %s: %s", receiver, r.Err)
+		receiver := vote.Sender
+		gr.Go(func() error {
+			rs, err := f.RClient.OverwriteObjects(ctx, receiver, f.class, shard, query)
+			if err != nil {
+				return fmt.Errorf("node %q could not repair objects: %w", receiver, err)
 			}
-		}
+			for _, r := range rs {
+				if r.Err != "" {
+					return fmt.Errorf("object changed in the meantime on node %s: %s", receiver, r.Err)
+				}
+			}
+			return nil
+		})
 	}
+	err := gr.Wait()
 	if nDeletions > 0 {
 		return result, errConflictExistOrDeleted
 	}
 
-	return result, nil
+	return result, err
 }
 
 // iTuple tuple of indices used to identify a unique object
@@ -519,7 +542,8 @@ func (f *Finder) readExistence(ctx context.Context,
 					maxCount = votes[i].ack
 				}
 				if maxCount >= st.Level {
-					resultCh <- result[bool]{!votes[i].o.Deleted, nil}
+					exists := !votes[i].o.Deleted && votes[i].o.UpdateTime != 0
+					resultCh <- result[bool]{exists, nil}
 					return
 				}
 			}
@@ -566,22 +590,26 @@ func (f *Finder) repairExist(ctx context.Context, shard string, id strfmt.UUID,
 	if resp.UpdateTime() != lastUTime {
 		return false, fmt.Errorf("fetch new state from %s: %w, %v", winner.sender, errConflictObjectChanged, err)
 	}
-
-	for _, c := range votes { // repair
-		if c.UTime != lastUTime {
-			updates := []*objects.VObject{{
+	gr, ctx := errgroup.WithContext(ctx)
+	for _, vote := range votes { // repair
+		if vote.UTime == lastUTime {
+			continue
+		}
+		vote := vote
+		gr.Go(func() error {
+			ups := []*objects.VObject{{
 				LatestObject:    &resp.Object.Object,
-				StaleUpdateTime: c.UTime,
-				// Version:         0,
+				StaleUpdateTime: vote.UTime,
 			}}
-			resp, err := f.RClient.OverwriteObjects(ctx, c.sender, f.class, shard, updates)
+			resp, err := f.RClient.OverwriteObjects(ctx, vote.sender, f.class, shard, ups)
 			if err != nil {
-				return false, fmt.Errorf("node %q could not repair object: %w", c.sender, err)
+				return fmt.Errorf("node %q could not repair object: %w", vote.sender, err)
 			}
 			if len(resp) > 0 && resp[0].Err != "" && resp[0].UpdateTime != lastUTime {
-				return false, fmt.Errorf("overwrite %w %s: %s", errConflictObjectChanged, c.sender, resp[0].Err)
+				return fmt.Errorf("overwrite %w %s: %s", errConflictObjectChanged, vote.sender, resp[0].Err)
 			}
-		}
+			return nil
+		})
 	}
-	return !resp.Deleted, nil
+	return !resp.Deleted, gr.Wait()
 }
diff --git a/usecases/replica/remote_replica_incoming.go b/usecases/replica/remote_replica_incoming.go
index b5601d5ac8..5a0c70a516 100644
--- a/usecases/replica/remote_replica_incoming.go
+++ b/usecases/replica/remote_replica_incoming.go
@@ -42,8 +42,6 @@ type RemoteReplicaIncomingRepo interface {
 	// Read endpoints
 	FetchObject(ctx context.Context, indexName,
 		shardName string, id strfmt.UUID) (objects.Replica, error)
-	DoesExist(ctx context.Context, class,
-		shardName string, id strfmt.UUID) (objects.Replica, error)
 	FetchObjects(ctx context.Context, class,
 		shardName string, ids []strfmt.UUID) ([]objects.Replica, error)
 	DigestObjects(ctx context.Context, class, shardName string,
@@ -120,12 +118,6 @@ func (rri *RemoteReplicaIncoming) FetchObject(ctx context.Context,
 	return rri.repo.FetchObject(ctx, indexName, shardName, id)
 }
 
-func (rri *RemoteReplicaIncoming) DoesExist(ctx context.Context,
-	indexName, shardName string, id strfmt.UUID,
-) (objects.Replica, error) {
-	return rri.repo.DoesExist(ctx, indexName, shardName, id)
-}
-
 func (rri *RemoteReplicaIncoming) FetchObjects(ctx context.Context,
 	indexName, shardName string, ids []strfmt.UUID,
 ) ([]objects.Replica, error) {


Test patch:
diff --git a/adapters/handlers/rest/handlers_objects_test.go b/adapters/handlers/rest/handlers_objects_test.go
index c30c24de4e..da96b70a51 100644
--- a/adapters/handlers/rest/handlers_objects_test.go
+++ b/adapters/handlers/rest/handlers_objects_test.go
@@ -1047,7 +1047,9 @@ type fakeManager struct {
 	deleteRefErr       *uco.Error
 }
 
-func (f *fakeManager) HeadObject(context.Context, *models.Principal, string, strfmt.UUID) (bool, *uco.Error) {
+func (f *fakeManager) HeadObject(context.Context, *models.Principal,
+	string, strfmt.UUID, *additional.ReplicationProperties,
+) (bool, *uco.Error) {
 	return f.headObjectReturn, f.headObjectErr
 }
 
@@ -1057,7 +1059,9 @@ func (f *fakeManager) AddObject(_ context.Context, _ *models.Principal,
 	return object, nil
 }
 
-func (f *fakeManager) ValidateObject(_ context.Context, _ *models.Principal, _ *models.Object) error {
+func (f *fakeManager) ValidateObject(_ context.Context, _ *models.Principal,
+	_ *models.Object, _ *additional.ReplicationProperties,
+) error {
 	panic("not implemented") // TODO: Implement
 }
 
diff --git a/adapters/repos/db/clusterintegrationtest/cluster_integration_test.go b/adapters/repos/db/clusterintegrationtest/cluster_integration_test.go
index 03987def4e..229a909c52 100644
--- a/adapters/repos/db/clusterintegrationtest/cluster_integration_test.go
+++ b/adapters/repos/db/clusterintegrationtest/cluster_integration_test.go
@@ -155,7 +155,7 @@ func testDistributed(t *testing.T, dirName string, batch bool) {
 		for _, obj := range data {
 			node := nodes[rand.Intn(len(nodes))]
 
-			ok, err := node.repo.Exists(context.Background(), distributedClass, obj.ID)
+			ok, err := node.repo.Exists(context.Background(), distributedClass, obj.ID, nil)
 			require.Nil(t, err)
 			assert.True(t, ok)
 		}
@@ -614,7 +614,7 @@ func testDistributed(t *testing.T, dirName string, batch bool) {
 			}
 
 			node := nodes[rand.Intn(len(nodes))]
-			actual, err := node.repo.Exists(context.Background(), distributedClass, obj.ID)
+			actual, err := node.repo.Exists(context.Background(), distributedClass, obj.ID, nil)
 			require.Nil(t, err)
 			assert.Equal(t, expected, actual)
 		}
diff --git a/adapters/repos/db/crud_integration_test.go b/adapters/repos/db/crud_integration_test.go
index dd93dc8c59..e78272cb97 100644
--- a/adapters/repos/db/crud_integration_test.go
+++ b/adapters/repos/db/crud_integration_test.go
@@ -117,7 +117,7 @@ func TestCRUD(t *testing.T) {
 	thingID := strfmt.UUID("a0b55b05-bc5b-4cc9-b646-1452d1390a62")
 
 	t.Run("validating that the thing doesn't exist prior", func(t *testing.T) {
-		ok, err := repo.Exists(context.Background(), "TheBestThingClass", thingID)
+		ok, err := repo.Exists(context.Background(), "TheBestThingClass", thingID, nil)
 		require.Nil(t, err)
 		assert.False(t, ok)
 	})
@@ -169,7 +169,7 @@ func TestCRUD(t *testing.T) {
 	})
 
 	t.Run("validating that the thing exists now", func(t *testing.T) {
-		ok, err := repo.Exists(context.Background(), "TheBestThingClass", thingID)
+		ok, err := repo.Exists(context.Background(), "TheBestThingClass", thingID, nil)
 		require.Nil(t, err)
 		assert.True(t, ok)
 	})
diff --git a/adapters/repos/db/lsmkv/memtable_roaring_set_test.go b/adapters/repos/db/lsmkv/memtable_roaring_set_test.go
index d7e1f8e2d0..f0ba7ea870 100644
--- a/adapters/repos/db/lsmkv/memtable_roaring_set_test.go
+++ b/adapters/repos/db/lsmkv/memtable_roaring_set_test.go
@@ -12,6 +12,7 @@
 package lsmkv
 
 import (
+	"path"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -20,8 +21,12 @@ import (
 )
 
 func TestMemtableRoaringSet(t *testing.T) {
+	memPath := func() string {
+		return path.Join(t.TempDir(), "fake")
+	}
+
 	t.Run("inserting individual entries", func(t *testing.T) {
-		m, err := newMemtable("fake", StrategyRoaringSet, 0, nil)
+		m, err := newMemtable(memPath(), StrategyRoaringSet, 0, nil)
 		require.Nil(t, err)
 
 		key1, key2 := []byte("key1"), []byte("key2")
@@ -48,7 +53,7 @@ func TestMemtableRoaringSet(t *testing.T) {
 	})
 
 	t.Run("inserting lists", func(t *testing.T) {
-		m, err := newMemtable("fake", StrategyRoaringSet, 0, nil)
+		m, err := newMemtable(memPath(), StrategyRoaringSet, 0, nil)
 		require.Nil(t, err)
 
 		key1, key2 := []byte("key1"), []byte("key2")
@@ -73,7 +78,7 @@ func TestMemtableRoaringSet(t *testing.T) {
 	})
 
 	t.Run("inserting bitmaps", func(t *testing.T) {
-		m, err := newMemtable("fake", StrategyRoaringSet, 0, nil)
+		m, err := newMemtable(memPath(), StrategyRoaringSet, 0, nil)
 		require.Nil(t, err)
 
 		key1, key2 := []byte("key1"), []byte("key2")
@@ -100,7 +105,7 @@ func TestMemtableRoaringSet(t *testing.T) {
 	})
 
 	t.Run("removing individual entries", func(t *testing.T) {
-		m, err := newMemtable("fake", StrategyRoaringSet, 0, nil)
+		m, err := newMemtable(memPath(), StrategyRoaringSet, 0, nil)
 		require.Nil(t, err)
 
 		key1, key2 := []byte("key1"), []byte("key2")
@@ -121,7 +126,7 @@ func TestMemtableRoaringSet(t *testing.T) {
 	})
 
 	t.Run("removing lists", func(t *testing.T) {
-		m, err := newMemtable("fake", StrategyRoaringSet, 0, nil)
+		m, err := newMemtable(memPath(), StrategyRoaringSet, 0, nil)
 		require.Nil(t, err)
 
 		key1, key2 := []byte("key1"), []byte("key2")
@@ -146,7 +151,7 @@ func TestMemtableRoaringSet(t *testing.T) {
 	})
 
 	t.Run("removing bitmaps", func(t *testing.T) {
-		m, err := newMemtable("fake", StrategyRoaringSet, 0, nil)
+		m, err := newMemtable(memPath(), StrategyRoaringSet, 0, nil)
 		require.Nil(t, err)
 
 		key1, key2 := []byte("key1"), []byte("key2")
@@ -171,7 +176,7 @@ func TestMemtableRoaringSet(t *testing.T) {
 	})
 
 	t.Run("adding/removing bitmaps", func(t *testing.T) {
-		m, err := newMemtable("fake", StrategyRoaringSet, 0, nil)
+		m, err := newMemtable(memPath(), StrategyRoaringSet, 0, nil)
 		require.Nil(t, err)
 
 		key1, key2 := []byte("key1"), []byte("key2")
diff --git a/test/acceptance/replication/crud_ops.go b/test/acceptance/replication/crud_ops.go
index 69c54d8f50..78410592ba 100644
--- a/test/acceptance/replication/crud_ops.go
+++ b/test/acceptance/replication/crud_ops.go
@@ -58,6 +58,11 @@ func getObject(t *testing.T, host, class string, id strfmt.UUID) (*models.Object
 	return helper.GetObject(t, class, id)
 }
 
+func objectExistsCL(t *testing.T, host, class string, id strfmt.UUID, cl replica.ConsistencyLevel) (bool, error) {
+	helper.SetupClient(host)
+	return helper.ObjectExistsCL(t, class, id, cl)
+}
+
 func getObjectCL(t *testing.T, host, class string, id strfmt.UUID, cl replica.ConsistencyLevel) (*models.Object, error) {
 	helper.SetupClient(host)
 	return helper.GetObjectCL(t, class, id, cl)
@@ -73,6 +78,11 @@ func patchObject(t *testing.T, host string, patch *models.Object) {
 	helper.PatchObject(t, patch)
 }
 
+func updateObjectCL(t *testing.T, host string, obj *models.Object, cl replica.ConsistencyLevel) {
+	helper.SetupClient(host)
+	helper.UpdateObjectCL(t, obj, cl)
+}
+
 func addReferences(t *testing.T, host string, refs []*models.BatchReference) {
 	helper.SetupClient(host)
 	helper.AddReferences(t, refs)
diff --git a/test/acceptance/replication/read_repair_test.go b/test/acceptance/replication/read_repair_test.go
index 6676db9d72..c583192969 100644
--- a/test/acceptance/replication/read_repair_test.go
+++ b/test/acceptance/replication/read_repair_test.go
@@ -41,7 +41,6 @@ func readRepair(t *testing.T) {
 		}
 	}()
 
-	// create classes
 	helper.SetupClient(compose.GetWeaviate().URI())
 	paragraphClass := articles.ParagraphsClass()
 	articleClass := articles.ArticlesClass()
@@ -57,8 +56,6 @@ func readRepair(t *testing.T) {
 		helper.CreateClass(t, articleClass)
 	})
 
-	// create objects
-
 	t.Run("insert paragraphs", func(t *testing.T) {
 		batch := make([]*models.Object, len(paragraphIDs))
 		for i, id := range paragraphIDs {
@@ -81,7 +78,6 @@ func readRepair(t *testing.T) {
 		createObjects(t, compose.GetWeaviateNode2().URI(), batch)
 	})
 
-	// take a node down
 	t.Run("stop node 2", func(t *testing.T) {
 		stopNode(ctx, t, compose, compose.GetWeaviateNode2().Name())
 		time.Sleep(10 * time.Second)
@@ -96,34 +92,66 @@ func readRepair(t *testing.T) {
 		Vector: []float32{1, 2, 3, 4, 5},
 	}
 
-	// add new object on remaining node
 	t.Run("add new object to node one", func(t *testing.T) {
 		createObjectCL(t, compose.GetWeaviate().URI(), &repairObj, replica.One)
 	})
 
-	// restart downed node
 	t.Run("restart node 2", func(t *testing.T) {
 		err = compose.Start(ctx, compose.GetWeaviateNode2().Name())
 		require.Nil(t, err)
 	})
 
-	// read request object
 	t.Run("run fetch to trigger read repair", func(t *testing.T) {
-		resp, err := getObject(t, compose.GetWeaviate().URI(), repairObj.Class, repairObj.ID)
+		_, err := getObject(t, compose.GetWeaviate().URI(), repairObj.Class, repairObj.ID)
 		require.Nil(t, err)
-		// TODO: replace with resp assertions
-		t.Logf("resp: %+v", resp)
 	})
 
-	// assert that repair was made
-	t.Run("assert read repair was made", func(t *testing.T) {
+	t.Run("assert new object read repair was made", func(t *testing.T) {
 		stopNode(ctx, t, compose, compose.GetWeaviate().Name())
 		time.Sleep(10 * time.Second)
 
-		resp, err := getObjectCL(t, compose.GetWeaviateNode2().URI(), repairObj.Class, repairObj.ID, replica.One)
+		resp, err := getObjectCL(t, compose.GetWeaviateNode2().URI(),
+			repairObj.Class, repairObj.ID, replica.One)
 		require.Nil(t, err)
 		assert.Equal(t, repairObj.ID, resp.ID)
 		assert.Equal(t, repairObj.Class, resp.Class)
 		assert.EqualValues(t, repairObj.Properties, resp.Properties)
 	})
+
+	replaceObj := repairObj
+	replaceObj.Properties = map[string]interface{}{
+		"contents": "this paragraph was replaced",
+	}
+
+	t.Run("replace object", func(t *testing.T) {
+		updateObjectCL(t, compose.GetWeaviateNode2().URI(), &replaceObj, replica.One)
+	})
+
+	t.Run("restart node 1", func(t *testing.T) {
+		restartNode1(ctx, t, compose)
+	})
+
+	t.Run("run exists to trigger read repair", func(t *testing.T) {
+		exists, err := objectExistsCL(t, compose.GetWeaviateNode2().URI(),
+			replaceObj.Class, replaceObj.ID, replica.All)
+		require.Nil(t, err)
+		require.True(t, exists)
+	})
+
+	t.Run("assert updated object read repair was made", func(t *testing.T) {
+		stopNode(ctx, t, compose, compose.GetWeaviateNode2().Name())
+		time.Sleep(10 * time.Second)
+
+		exists, err := objectExistsCL(t, compose.GetWeaviate().URI(),
+			replaceObj.Class, replaceObj.ID, replica.One)
+		require.Nil(t, err)
+		require.True(t, exists)
+
+		resp, err := getObjectCL(t, compose.GetWeaviate().URI(),
+			repairObj.Class, repairObj.ID, replica.One)
+		require.Nil(t, err)
+		assert.Equal(t, replaceObj.ID, resp.ID)
+		assert.Equal(t, replaceObj.Class, resp.Class)
+		assert.EqualValues(t, replaceObj.Properties, resp.Properties)
+	})
 }
diff --git a/test/helper/objects.go b/test/helper/objects.go
index bb1ecc29ee..715c66ee60 100644
--- a/test/helper/objects.go
+++ b/test/helper/objects.go
@@ -84,6 +84,14 @@ func UpdateObject(t *testing.T, object *models.Object) {
 	AssertRequestOk(t, resp, err, nil)
 }
 
+func UpdateObjectCL(t *testing.T, object *models.Object, cl replica.ConsistencyLevel) {
+	cls := string(cl)
+	params := objects.NewObjectsClassPutParams().WithClassName(object.Class).
+		WithID(object.ID).WithBody(object).WithConsistencyLevel(&cls)
+	resp, err := Client(t).Objects.ObjectsClassPut(params, nil)
+	AssertRequestOk(t, resp, err, nil)
+}
+
 func PatchObject(t *testing.T, object *models.Object) {
 	params := objects.NewObjectsPatchParams().WithID(object.ID).WithBody(object)
 	resp, err := Client(t).Objects.ObjectsPatch(params, nil)
diff --git a/test/helper/objects_assertions.go b/test/helper/objects_assertions.go
index f4d88b9741..8a0cf08ada 100644
--- a/test/helper/objects_assertions.go
+++ b/test/helper/objects_assertions.go
@@ -12,6 +12,7 @@
 package helper
 
 import (
+	"net/http"
 	"testing"
 
 	"github.com/go-openapi/strfmt"
@@ -127,6 +128,17 @@ func GetObjectCL(t *testing.T, class string, uuid strfmt.UUID,
 	return getResp.Payload, nil
 }
 
+func ObjectExistsCL(t *testing.T, class string, id strfmt.UUID, cl replica.ConsistencyLevel) (bool, error) {
+	cls := string(cl)
+	req := objects.NewObjectsClassHeadParams().
+		WithClassName(class).WithID(id).WithConsistencyLevel(&cls)
+	resp, err := Client(t).Objects.ObjectsClassHead(req, nil)
+	if err != nil {
+		return false, err
+	}
+	return resp.IsCode(http.StatusNoContent), nil
+}
+
 func GetObjectFromNode(t *testing.T, class string, uuid strfmt.UUID, nodename string) (*models.Object, error) {
 	req := objects.NewObjectsClassGetParams().WithID(uuid)
 	if class != "" {
diff --git a/usecases/objects/fakes_for_test.go b/usecases/objects/fakes_for_test.go
index c7c7c97be3..733d735e5d 100644
--- a/usecases/objects/fakes_for_test.go
+++ b/usecases/objects/fakes_for_test.go
@@ -140,7 +140,7 @@ type fakeVectorRepo struct {
 }
 
 func (f *fakeVectorRepo) Exists(ctx context.Context, class string,
-	id strfmt.UUID,
+	id strfmt.UUID, repl *additional.ReplicationProperties,
 ) (bool, error) {
 	args := f.Called(class, id)
 	return args.Bool(0), args.Error(1)
diff --git a/usecases/objects/head_test.go b/usecases/objects/head_test.go
index fbb58b9397..6b568a3684 100644
--- a/usecases/objects/head_test.go
+++ b/usecases/objects/head_test.go
@@ -78,7 +78,7 @@ func Test_HeadObject(t *testing.T) {
 		if tc.authErr == nil && tc.lockErr == nil {
 			m.repo.On("Exists", tc.class, id).Return(tc.mockedOk, tc.mockedErr).Once()
 		}
-		ok, err := m.Manager.HeadObject(context.Background(), nil, tc.class, id)
+		ok, err := m.Manager.HeadObject(context.Background(), nil, tc.class, id, nil)
 		code := 0
 		if err != nil {
 			code = err.Code
diff --git a/usecases/objects/validation/fakes_for_test.go b/usecases/objects/validation/fakes_for_test.go
index 73f46a78aa..dc0dacfdbe 100644
--- a/usecases/objects/validation/fakes_for_test.go
+++ b/usecases/objects/validation/fakes_for_test.go
@@ -15,6 +15,7 @@ import (
 	"context"
 
 	"github.com/go-openapi/strfmt"
+	"github.com/weaviate/weaviate/entities/additional"
 	"github.com/weaviate/weaviate/entities/models"
 	"github.com/weaviate/weaviate/entities/schema"
 )
@@ -41,6 +42,6 @@ func testSchema() schema.Schema {
 	}
 }
 
-func fakeExists(context.Context, string, strfmt.UUID) (bool, error) {
+func fakeExists(context.Context, string, strfmt.UUID, *additional.ReplicationProperties) (bool, error) {
 	return true, nil
 }
diff --git a/usecases/objects/validation/phone_numbers_test.go b/usecases/objects/validation/phone_numbers_test.go
index 65b1a0fbe0..dc612a9cca 100644
--- a/usecases/objects/validation/phone_numbers_test.go
+++ b/usecases/objects/validation/phone_numbers_test.go
@@ -159,7 +159,7 @@ func TestPropertyOfTypePhoneNumberValidation(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			config := &config.WeaviateConfig{}
-			validator := New(fakeExists, config)
+			validator := New(fakeExists, config, nil)
 
 			obj := &models.Object{
 				Class: "Person",
diff --git a/usecases/replica/finder_test.go b/usecases/replica/finder_test.go
index b6bc1de789..8f83a51dd4 100644
--- a/usecases/replica/finder_test.go
+++ b/usecases/replica/finder_test.go
@@ -1376,7 +1376,7 @@ func TestFinderGetAllWithConsistencyLevelAll(t *testing.T) {
 		got, err := finder.GetAll(ctx, All, shard, ids)
 		assert.ErrorContains(t, err, msgCLevel)
 		assert.ErrorIs(t, err, errConflictExistOrDeleted)
-		assert.Equal(t, result, got)
+		assert.Equal(t, nilObjects, got)
 	})
 }