Repository: googleapis/google-resumable-media-python

Issue:
After retry limit reached, ResumableUpload.transmit_next_chunk propagates exception with invalid==False
The code below with comments probably best describes the issue... it shows two workarounds. A third workaround (not shown) might be to look for a retry timeout exception and assume the ResumableUpload is invalid despite it being in a recoverable state in my various tests. I'm wondering if a call to _make_invalid upon final retry limit reached is reasonable, or perhaps documenting implied invalidity after retry limit though the latter would seem odd since a retry limit having been reached is not equivalent to a non-recoverable state. 

```
upload = ResumableUpload(upload_url, chunk_size)
response = upload.initiate(transport=session, 
                            stream=stream, 
                            metadata=metadata, 
                            content_type="application/octet-stream", 
                            stream_final=False)
while not upload.finished:
    try:
        pos_before_xmit = stream.tell()
        upload.transmit_next_chunk(session)
    except Exception as ex:
        bytes_attempted = stream.tell() - pos_before_xmit
        """ 
        #
        # I first tried the following which skips recover
        # and jumps back to transmit_next_chunk above.
        # Without this, transmit_next_chunk fails. This
        # seems somewhat equivalent to continuing
        # above with retry reset.
        #
        if not upload.invalid:
            stream.seek(pos_before_xmit, io.SEEK_SET)
        else:
            r = upload.recover(session)
        """
        #
        # After looking at internals, I tried this which
        # also worked. Without the _make_invalid call,
        # recover fails and the stream is not rewound,
        # causing transmit_next_chunk to fail. 
        #
        if not upload.invalid:
            upload._make_invalid() # not really supposed to be called by users.
        r = upload.recover(session)
        # etc.
```

An aside, thanks to authors/contribs who created ResumableUpload.

Discussion:

Patch:
diff --git a/google/resumable_media/_upload.py b/google/resumable_media/_upload.py
index b4eac56..c681cc0 100644
--- a/google/resumable_media/_upload.py
+++ b/google/resumable_media/_upload.py
@@ -794,14 +794,8 @@ class ResumableUpload(UploadBase):
             The headers **do not** incorporate the ``_headers`` on the
             current instance.
 
-        Raises:
-            ValueError: If the current upload is not in an invalid state.
-
         .. _sans-I/O: https://sans-io.readthedocs.io/
         """
-        if not self.invalid:
-            raise ValueError("Upload is not in invalid state, no need to recover.")
-
         headers = {_helpers.CONTENT_RANGE_HEADER: "bytes */*"}
         return _PUT, self.resumable_url, None, headers
 
diff --git a/google/resumable_media/requests/upload.py b/google/resumable_media/requests/upload.py
index b565d7a..0649001 100644
--- a/google/resumable_media/requests/upload.py
+++ b/google/resumable_media/requests/upload.py
@@ -517,14 +517,15 @@ class ResumableUpload(_request_helpers.RequestsMixin, _upload.ResumableUpload):
         )
 
     def recover(self, transport):
-        """Recover from a failure.
-
-        This method should be used when a :class:`ResumableUpload` is in an
-        :attr:`~ResumableUpload.invalid` state due to a request failure.
+        """Recover from a failure and check the status of the current upload.
 
         This will verify the progress with the server and make sure the
         current upload is in a valid state before :meth:`transmit_next_chunk`
-        can be used again.
+        can be used again. See https://cloud.google.com/storage/docs/performing-resumable-uploads#status-check
+        for more information.
+
+        This method can be used when a :class:`ResumableUpload` is in an
+        :attr:`~ResumableUpload.invalid` state due to a request failure.
 
         Args:
             transport (~requests.Session): A ``requests`` object which can


Test patch:
diff --git a/tests/unit/test__upload.py b/tests/unit/test__upload.py
index 110ed77..db943d9 100644
--- a/tests/unit/test__upload.py
+++ b/tests/unit/test__upload.py
@@ -951,8 +951,13 @@ class TestResumableUpload(object):
         upload = _upload.ResumableUpload(RESUMABLE_URL, ONE_MB)
         assert not upload.invalid
 
-        with pytest.raises(ValueError):
-            upload._prepare_recover_request()
+        method, url, payload, headers = upload._prepare_recover_request()
+        assert method == "PUT"
+        assert url == upload.resumable_url
+        assert payload is None
+        assert headers == {"content-range": "bytes */*"}
+        # Make sure headers are untouched.
+        assert upload._headers == {}
 
     def test__prepare_recover_request(self):
         upload = _upload.ResumableUpload(RESUMABLE_URL, ONE_MB)