Async retry fix + file handle improved

telebot/apihelper.py

@@ -137,6 +137,11 @@ def _make_request(token, method_name, method='get', params=None, files=None):
     elif RETRY_ON_ERROR and RETRY_ENGINE == 1:
         got_result = False
         current_try = 0
+        file_streams = (
+            value[1] if isinstance(value, (tuple, list)) and len(value) >= 2 else value
+            for value in (files or {}).values()
+        )
+        file_positions = util._get_retryable_file_positions(file_streams)
         while not got_result and current_try<MAX_RETRIES-1:
             current_try+=1
             try:
@@ -146,12 +151,18 @@ def _make_request(token, method_name, method='get', params=None, files=None):
                 got_result = True
             except HTTPError:
                 logger.debug("HTTP Error on {0} method (Try #{1})".format(method_name, current_try))
+                if file_positions is None or not util._rewind_file_positions(file_positions):
+                    raise
                 time.sleep(RETRY_TIMEOUT)
             except ConnectionError:
                 logger.debug("Connection Error on {0} method (Try #{1})".format(method_name, current_try))
+                if file_positions is None or not util._rewind_file_positions(file_positions):
+                    raise
                 time.sleep(RETRY_TIMEOUT)
             except Timeout:
                 logger.debug("Timeout Error on {0} method (Try #{1})".format(method_name, current_try))
+                if file_positions is None or not util._rewind_file_positions(file_positions):
+                    raise
                 time.sleep(RETRY_TIMEOUT)
         if not got_result:
             result = _get_req_session().request(

telebot/asyncio_helper.py

@@ -26,6 +26,8 @@
 
 REQUEST_TIMEOUT = 300
 MAX_RETRIES = 3
+RETRY_ON_ERROR = False
+RETRY_TIMEOUT = 2
 
 REQUEST_LIMIT = 50
 
@@ -66,6 +68,7 @@ async def get_session(self):
 
 session_manager = SessionManager()
 
+
 async def _process_request(token, url, method='get', params=None, files=None, **kwargs):
     # Let's resolve all timeout parameters.
     # getUpdates parameter may contain 2 parameters: request_timeout & timeout.
@@ -86,32 +89,45 @@ async def _process_request(token, url, method='get', params=None, files=None, **
     request_timeout = REQUEST_TIMEOUT if request_timeout is None else request_timeout
 
 
-    # Preparing data by adding all parameters and files to FormData
-    params = _prepare_data(params, files)
-
     timeout = aiohttp.ClientTimeout(total=request_timeout)
-    got_result = False
-    current_try=0
+    max_attempts = max(1, MAX_RETRIES if RETRY_ON_ERROR else 1)
+    file_streams = []
+    for value in (files or {}).values():
+        if isinstance(value, tuple):
+            if len(value) != 2:
+                continue
+            value = value[1]
+        if isinstance(value, types.InputFile):
+            value = value.file
+        file_streams.append(value)
+    file_positions = util._get_retryable_file_positions(file_streams)
+    last_error = None
     session = await session_manager.get_session()
-    while not got_result and current_try<MAX_RETRIES-1:
-        current_try +=1
+    for current_try in range(1, max_attempts + 1):
+        # FormData is consumed when aiohttp builds a multipart request, so it
+        # must be created again for each retry.
+        request_data = _prepare_data(params, files)
         try:
-            async with session.request(method=method, url=API_URL.format(token, url), data=params, timeout=timeout, proxy=proxy) as resp:
-                got_result = True
+            async with session.request(method=method, url=API_URL.format(token, url), data=request_data, timeout=timeout, proxy=proxy) as resp:
                 logger.debug("Request: method={0} url={1} params={2} files={3} request_timeout={4} current_try={5}".format(method, url, params, files, request_timeout, current_try).replace(token, token.split(':')[0] + ":{TOKEN}"))
 
                 json_result = await _check_result(url, resp)
                 if json_result:
                     return json_result['result']
+                return None
         except (ApiTelegramException,ApiInvalidJSONException, ApiHTTPException) as e:
             raise e
-        except aiohttp.ClientError as e:
-            logger.error('Aiohttp ClientError: {0}'.format(e.__class__.__name__))
-        except Exception as e:
-            logger.error(f'Unknown error: {e.__class__.__name__}')
-        if not got_result:
-            raise RequestTimeout("Request timeout. Request: method={0} url={1} params={2} files={3} request_timeout={4}".format(method, url, params, files, request_timeout, current_try))
-    return None
+        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
+            last_error = e
+            logger.error('Aiohttp request error: {0} (try #{1})'.format(e.__class__.__name__, current_try))
+
+            if current_try == max_attempts:
+                break
+            if file_positions is None or not util._rewind_file_positions(file_positions):
+                break
+            await asyncio.sleep(RETRY_TIMEOUT)
+
+    raise RequestTimeout("Request timeout. Request: method={0} url={1} params={2} files={3} request_timeout={4} current_try={5}".format(method, url, params, files, request_timeout, current_try)) from last_error
 
 def _prepare_file(obj):
     """

telebot/util.py

@@ -536,6 +536,41 @@ def per_thread(key, construct_value, reset=False):
     return getattr(thread_local, key)
 
 
+def _get_retryable_file_positions(file_streams):
+    """Return stream positions, or ``None`` when one cannot be rewound.
+
+    Multipart encoders consume file streams while preparing a request. A retry
+    must restore every stream to its position before the first attempt;
+    otherwise, it can upload an empty or truncated file. The caller supplies
+    already-extracted file-like objects so this helper remains independent of
+    any HTTP client's multipart value format.
+    """
+    positions = []
+    for file_stream in file_streams:
+        if not hasattr(file_stream, 'read'):
+            continue
+        try:
+            positions.append((file_stream, file_stream.tell()))
+        except (AttributeError, OSError, ValueError):
+            return None
+    return positions
+
+
+def _rewind_file_positions(positions):
+    """Restore streams to positions returned by :func:`_get_retryable_file_positions`.
+
+    A ``False`` result means that at least one stream cannot safely be
+    restored. In that case, the caller must not retry the multipart request,
+    because doing so could send an empty or truncated upload.
+    """
+    try:
+        for file_stream, position in positions:
+            file_stream.seek(position)
+    except (AttributeError, OSError, ValueError):
+        return False
+    return True
+
+
 def deprecated(warn: bool = True, alternative: Optional[Callable] = None, deprecation_text=None):
     """
     Use this decorator to mark functions as deprecated.