test corner cases

Author: vgvoleg

tests/topics/test_topic_transactions.py

@@ -65,7 +65,7 @@ async def callee(tx: ydb.aio.QueryTxContext):
         with pytest.raises(asyncio.TimeoutError):
             await wait_for(topic_reader.receive_message(), 0.1)
 
-    async def test_no_msg_writter_in_error_case(
+    async def test_no_msg_written_in_error_case(
         self, driver: ydb.aio.Driver, topic_path, topic_reader: ydb.TopicReaderAsyncIO
     ):
         async with ydb.aio.QuerySessionPool(driver) as pool:
@@ -82,6 +82,29 @@ async def callee(tx: ydb.aio.QueryTxContext):
         with pytest.raises(asyncio.TimeoutError):
             await wait_for(topic_reader.receive_message(), 0.1)
 
+    async def test_msg_written_exactly_once_with_retries(
+        self, driver: ydb.aio.Driver, topic_path, topic_reader: ydb.TopicReaderAsyncIO
+    ):
+        error_raised = False
+        async with ydb.aio.QuerySessionPool(driver) as pool:
+
+            async def callee(tx: ydb.aio.QueryTxContext):
+                nonlocal error_raised
+                tx_writer = driver.topic_client.tx_writer(tx, topic_path)
+                await tx_writer.write(ydb.TopicWriterMessage(data="123".encode()))
+
+                if not error_raised:
+                    error_raised = True
+                    raise ydb.issues.Unavailable("some retriable error")
+
+            await pool.retry_tx_async(callee)
+
+            msg = await wait_for(topic_reader.receive_message(), 0.1)
+            assert msg.data.decode() == "123"
+
+            with pytest.raises(asyncio.TimeoutError):
+                await wait_for(topic_reader.receive_message(), 0.1)
+
 
 class TestTopicTransactionalWriterSync:
     def test_commit(self, driver_sync: ydb.Driver, topic_path, topic_reader_sync: ydb.TopicReader):
@@ -110,7 +133,7 @@ def callee(tx: ydb.QueryTxContext):
         with pytest.raises(TimeoutError):
             topic_reader_sync.receive_message(timeout=0.1)
 
-    def test_no_msg_writter_in_error_case(
+    def test_no_msg_written_in_error_case(
         self, driver_sync: ydb.Driver, topic_path, topic_reader_sync: ydb.TopicReaderAsyncIO
     ):
         with ydb.QuerySessionPool(driver_sync) as pool:

ydb/_topic_writer/topic_writer_asyncio.py

@@ -191,7 +191,7 @@ async def _on_before_commit(self):
         await self.close()
 
     async def _on_before_rollback(self):
-        await self.close()
+        await self.close(flush=False)
 
 
 class WriterAsyncIOReconnector:
@@ -423,7 +423,7 @@ async def _connection_loop(self):
                 done.pop().result()  # need for raise exception - reason of stop task
             except issues.Error as err:
                 err_info = check_retriable_error(err, retry_settings, attempt)
-                if not err_info.is_retriable:
+                if not err_info.is_retriable or self._tx is not None:  # no retries in tx writer
                     self._stop(err)
                     return
 
@@ -586,7 +586,6 @@ async def _send_loop(self, writer: "WriterAsyncIOStream"):
 
             while True:
                 m = await self._new_messages.get()  # type: InternalMessage
-                print("NEW MESSAGE")
                 if m.seq_no > last_seq_no:
                     writer.write([m])
         except asyncio.CancelledError:
@@ -618,7 +617,6 @@ async def flush(self):
 
         # wait last message
         await asyncio.wait(self._messages_future)
-        print("ALL MESSAGES WERE SENT TO SERVER")
 
 
 class WriterAsyncIOStream:

ydb/_topic_writer/topic_writer_sync.py

@@ -164,4 +164,4 @@ def _on_before_commit(self):
         self.close()
 
     def _on_before_rollback(self):
-        self.close()
+        self.close(flush=False)