Fixes "the connection was closed by the remote peer" error

[Gapex]

SeekableRawReader reopens a connection after getting a IncompleteReadError

Motivation

• Fixes s3 open seek operation try read rest of file into buffer, which makes following read has timeout risk #362

The current SeekableRawReader sends HTTP request in the seek() method, and keeps an continuous connection with the remote peer. But if the SeekableRawReader does not read from the connection immediately after seek(), the remote peer may close the connection due to a timeout. Then, the SeekableRawReader attempts to read on a StreamingBody whose underlying connection had been closed, and gets a botocore.exceptions.IncompleteReadError, the following example shows the case:

from boto3 import resource

endpoints = [
    'http://xxxxxxxxxxxxx', # real s3
    'http://localhost:8000' # moto s3 server
]

real_s3_resource = resource('s3', endpoint_url=endpoints[0])
moto_s3_resource = resource('s3', endpoint_url=endpoints[1])

# Test read from a closed streaming body.

# For real s3.
obj = real_s3_resource.Object('gyp', 'images.msg')
body = obj.get()['Body']
print(type(body)) # <class 'botocore.response.StreamingBody'>
print(body.read(1)) # b'\xdd'
body.close() # Close the body.
try:
    body.read(1)
except Exception as e:
    print(type(e)) # <class 'botocore.exceptions.IncompleteReadError'>


# For moto s3 server.
moto_s3_resource.Bucket('gyp').create()
moto_s3_resource.Object('gyp', 'test-key').put(Body=b'123456')
obj = moto_s3_resource.Object('gyp', 'test-key')
body = obj.get()['Body']
print(type(body)) # <class 'botocore.response.StreamingBody'>
print(body.read(1)) # b'1'
body.close() # Close the body.
try:
    body.read(1)
except Exception as e:
    print(type(e)) # <class 'botocore.exceptions.IncompleteReadError'>

In the above code, the body is closed by client actively to stimulate connection closed by the remote peer.

In the PR, I intended to delay the action of sending HTTP request until read, the solution solved the problem of closed connection between a pair of seek() - read(), but introduced a similar problem between continuous read(), namely, the connection may also be closed by the remote peer after a read, if the second read occurs a long time later.

In this PR, I try to catch the IncompleteReadError in the read() directly, if connection is close, and IncompleteReadError is raised,

1. Catch the exception and dereference the current StreamingBody,
2. Call the seek() from the current offset to build a new connection with remote peer.
3. Finally, read from the new connection as required.

[Gapex]

@mpenkov

• The moto uses a file object to mock StreamingBody, and reading from a closed file can not actully raises botocore.exceptions.IncompleteReadError.
• So, I need to skip the test case in 'Mock Environment'.
• I notice that The travis unsets the SO_DISABLE_MOCKS env variable, thus all s3 tests were using moto.
• In order to test the modified part of code, a real connection maybe is needed.

from boto3 import resource

mocker = moto.mock_s3()
mocker.start()
r = resource('s3')
r.Bucket('bucket').create()
r.Object('bucket', 'key').put(Body=b'123456')
obj = r.Object('bucket', 'key')
body = obj.get()['Body']
print(body.read(1)) # b'1'
body.close()
try:
    print(body.read(1))
except Exception as e:
    print(type(e)) # <class 'ValueError'>
    print(e) # I/O operation on closed file.
mocker.stop()

[vuryleo]

I would think it's an issue related to moto. We may need to raise it there. :(

[mpenkov]

Let me know when this is no longer WIP and I'll have a closer look.

The overall direction looks right to me.

[mpenkov]

Typo:

Suggested change

	self._local_resoruce = boto3.resource('s3', endpoint_url='http://localhost:5000')
	self._local_resource = boto3.resource('s3', endpoint_url='http://localhost:5000')

[mpenkov]

@Gapex Thank you for your patience! This is a very welcome contribution. Keep them coming!