Use node writer response on set/delete resource (#1610)

* Use node writer response on set/delete resource

* use custom exception

nucliadb_node/nucliadb_node/pull.py

@@ -79,6 +79,10 @@
 )
 
 
+class IndexNodeError(Exception):
+    pass
+
+
 class ShardManager:
     schedule_delay_seconds = 30.0
 
@@ -307,10 +311,13 @@ async def subscription_worker(self, msg: Msg):
             msg, nats_consumer_settings.nats_ack_wait * 0.66
         ), sm.lock:
             try:
+                status = None
                 if pb.typemessage == TypeMessage.CREATION:
-                    await self.set_resource(pb)
+                    status = await self.set_resource(pb)
                 elif pb.typemessage == TypeMessage.DELETION:
-                    await self.delete_resource(pb)
+                    status = await self.delete_resource(pb)
+                if status is not None and status.status != OpStatus.Status.OK:
+                    raise IndexNodeError(status.detail)
                 sm.shard_changed_event()
             except AioRpcError as grpc_error:
                 if grpc_error.code() == StatusCode.NOT_FOUND:

nucliadb_node/nucliadb_node/tests/unit/test_pull.py

@@ -20,13 +20,13 @@
 import asyncio
 import tempfile
 from unittest import mock
-from unittest.mock import AsyncMock, MagicMock, Mock
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import pytest
 from nats.aio.client import Msg
-from nucliadb_protos.nodewriter_pb2 import IndexMessage, TypeMessage
+from nucliadb_protos.nodewriter_pb2 import IndexMessage, OpStatus, TypeMessage
 
-from nucliadb_node.pull import IndexedPublisher, ShardManager, Worker
+from nucliadb_node.pull import IndexedPublisher, IndexNodeError, ShardManager, Worker
 from nucliadb_node.settings import settings
 from nucliadb_utils import const
 
@@ -188,3 +188,13 @@ async def test_reconnected_cb(self, worker: Worker):
             assert worker.nc.jetstream().subscribe.call_count == 2
         finally:
             await worker.finalize()
+
+    @pytest.mark.asyncio
+    async def test_node_writer_errors_are_managed(self, worker: Worker):
+        status = OpStatus()
+        status.status = OpStatus.Status.ERROR
+        status.detail = "node writer error"
+        with patch("nucliadb_node.pull.Worker.set_resource", return_value=status):
+            msg = self.get_msg(seqid=1)
+            with pytest.raises(IndexNodeError):
+                await worker.subscription_worker(msg)