Drop inconsistent dimension errored messages (#2268)

* Drop inconsistent dimension errored messages

* Add more info to the log

* More log

nucliadb_sidecar/src/nucliadb_sidecar/indexer.py

@@ -77,6 +77,10 @@ class MetadataNotFoundError(IndexNodeError):
     pass
 
 
+class InconsistentDimensionsError(IndexNodeError):
+    pass
+
+
 class ShardNotFound(IndexNodeError):
     pass
 
@@ -398,7 +402,7 @@ async def _index_message(self, pb: IndexMessage):
     def _parse_op_status_errors(self, pb: IndexMessage, status: OpStatus):
         if status.status == OpStatus.Status.OK:
             return
-        if status.status == OpStatus.Status.ERROR and "Shard not found" in status.detail:
+        elif status.status == OpStatus.Status.ERROR and "Shard not found" in status.detail:
             logger.warning(
                 f"Shard does not exist {pb.shard}. This message will be ignored",
                 extra={
@@ -409,8 +413,10 @@ def _parse_op_status_errors(self, pb: IndexMessage, status: OpStatus):
                 },
             )
             raise ShardNotFound()
-        if status.status == OpStatus.Status.ERROR and "Missing resource metadata" in status.detail:
+        elif status.status == OpStatus.Status.ERROR and "Missing resource metadata" in status.detail:
             raise MetadataNotFoundError()
+        elif status.status == OpStatus.Status.ERROR and "Inconsistent dimensions" in status.detail:
+            raise InconsistentDimensionsError()
         raise IndexNodeError(status.detail)
 
     @contextlib.contextmanager
@@ -456,6 +462,18 @@ async def _set_resource(self, pb: IndexMessage) -> None:
                 },
             )
             return
+        except InconsistentDimensionsError:
+            logger.error(
+                "Trying to index vectors with an incorrect dimension! This should not happen!"
+                "We drop the message to not block the queue",
+                extra={
+                    "kbid": pb.kbid,
+                    "shard": pb.shard,
+                    "rid": pb.resource,
+                    "storage_key": pb.storage_key,
+                },
+            )
+            return
 
     async def _set_resource_from_storage(self, pb: IndexMessage) -> None:
         """
@@ -488,6 +506,18 @@ async def _set_resource_from_storage(self, pb: IndexMessage) -> None:
                 },
             )
             return
+        except InconsistentDimensionsError:
+            logger.error(
+                "Trying to index vectors with an incorrect dimension! This should not happen!"
+                "We drop the message to not block the queue",
+                extra={
+                    "kbid": pb.kbid,
+                    "shard": pb.shard,
+                    "rid": pb.resource,
+                    "storage_key": pb.storage_key,
+                },
+            )
+            return
 
     async def _delete_resource(self, pb: IndexMessage) -> None:
         shard_id = pb.shard

nucliadb_sidecar/tests/unit/test_indexer.py

@@ -314,13 +314,30 @@ async def test_node_writer_status_shard_not_found_errors_are_handled(
     ):
         status = OpStatus()
         status.status = OpStatus.Status.ERROR
-        status.detail = 'status: NotFound, message: "Shard not found: Shard'
+        status.detail = 'status: NotFound, message: "Shard not found: Shard"'
 
         writer.set_resource_from_storage.return_value = status
         pb = IndexMessage()
         pb.typemessage = TypeMessage.CREATION
         await indexer._index_message(pb)
 
+    @pytest.mark.asyncio
+    async def test_inconsistent_dimension_errors_are_handled(
+        self,
+        indexer: PriorityIndexer,
+        writer: AsyncMock,
+    ):
+        status = OpStatus()
+        status.status = OpStatus.Status.ERROR
+        status.detail = 'status: Error, message: "Inconsistent dimensions"'
+
+        writer.set_resource = AsyncMock(return_value=status)
+        writer.set_resource_from_storage = AsyncMock(return_value=status)
+
+        pb = IndexMessage()
+        pb.typemessage = TypeMessage.CREATION
+        await indexer._index_message(pb)
+
     @pytest.mark.asyncio
     async def test_node_writer_aio_rpc_errors_are_handled(
         self, indexer: PriorityIndexer, writer: AsyncMock