Repository: ska-sa/katdal

Issue:
mvftoms produced malformed MS (TiledStMan::headerFileGet: mismatch in #row)
Trying to use `mvftoms.py` to pull 32k data from the archive to IDIA. The connection frequently drops with:

```
socket.timeout: The read operation timed out
```

but when I restart it continues. It's got most of the way there after a couple of days (11 TB downloaded), but now I'm getting the error below. I get the same error when I try to access the MS using other software. Any tips on how to recover this gratefully received. I'll start the transfer again in the meantime.

Cheers.

```
/users/ianh/venv/katdal/lib/python3.8/site-packages/katdal/applycal.py:156: RuntimeWarning: invalid value encountered in reciprocal
  corrections.append(ComparableArrayWrapper(np.reciprocal(bp)))
/users/ianh/venv/katdal/lib/python3.8/site-packages/katdal/applycal.py:204: RuntimeWarning: invalid value encountered in reciprocal
  return np.reciprocal(smooth_gains)
The following calibration products will be applied: l1.K, l1.B, l1.G, l2.GPHASE
Per user request the following antennas will be selected: 'm000', 'm001', 'm002', 'm003', 'm004', 'm005', 'm006', 'm007', 'm008', 'm010', 'm012', 'm013', 'm014', 'm015', 'm016', 'm017', 'm018', 'm019', 'm020', 'm021', 'm022', 'm023', 'm024', 'm025', 'm026', 'm027', 'm028', 'm029', 'm030', 'm031', 'm032', 'm033', 'm034', 'm035', 'm036', 'm037', 'm038', 'm039', 'm041', 'm042', 'm043', 'm044', 'm045', 'm046', 'm047', 'm048', 'm049', 'm050', 'm051', 'm052', 'm053', 'm054', 'm055', 'm056', 'm057', 'm058', 'm059', 'm060', 'm061', 'm062', 'm063'
Per user request the following target fields will be selected: 'COSMOS_3'
Per user request the following scans will be dumped: 34, 8, 42, 14, 48, 22, 28
Extract MS for spw 0: centre frequency 1284000000 Hz
Will create MS output in 1622376680_sdp_l2.full.ms

#### Producing a full polarisation MS (HH,HV,VH,VV) ####


Using array as the reference antenna. All targets and scans will be based on this antenna.


Iterating through scans in dataset(s)...

Traceback (most recent call last):
  File "/users/ianh/venv/katdal/bin/mvftoms.py", line 916, in <module>
    main()
  File "/users/ianh/venv/katdal/bin/mvftoms.py", line 545, in main
    with ms_extra.open_table(ms_name, verbose=options.verbose) as t:
  File "/users/ianh/venv/katdal/lib/python3.8/site-packages/katdal/ms_extra.py", line 42, in open_table
    return tables.table(name, readonly=readonly, ack=verbose, **kwargs)
  File "/users/ianh/venv/katdal/lib/python3.8/site-packages/casacore/tables/table.py", line 373, in __init__
    Table.__init__(self, tabname, lockopt, opt)
RuntimeError: Table DataManager error: Internal error: TiledStMan::headerFileGet: mismatch in #row; expected 0, found 5313710
```

Hints from issue comments:
@ludwigschwardt: Sorry for your struggles! 

>  It's got most of the way there after a couple of days (11 TB downloaded),

Do you mean you have a partial MS after a couple of restarts, but now the MS is stuck after the latest restart?

The header probably has 5313710 rows on disk, but I'm still trying to understand why the DataManager expects an empty table...

@IanHeywood: > Sorry for your struggles!

No apologies necessary. I'm shamefully late when it comes to using this method, and it's already changing the game. I was looking for the 'help wanted' tag for this issue because I'm not sure it's actually a bug.

> Do you mean you have a partial MS after a couple of restarts, but now the MS is stuck after the latest restart?

The MS is 11 out of maybe 13 TB at this point, so it's annoyingly close. The socket.timeout thing is endemic though, there have been a lot of restarts for this particular observation and some others that seem to be OK. This method seems way faster than what I've been doing before, but I have to babysit it constantly because of the timeouts. I was looking to implement a (perhaps slightly dodgy) script that captures the mvftoms.py output and checks the last line for the time out error and then restarts the transfer if it's there. But this is a new one, it looks like the MS is fundamentally mangled...

> The header probably has 5313710 rows on disk, but I'm still trying to understand why the DataManager expects an empty table...

...and it it looks like some deep casacore lore that I don't know how to diagnose.

Cheers.@bennahugo: @IanHeywood - I know this is not a fix but if you look at the help options
you will see that you can select by scans and fields (rather than
downloading 11TiB in one go?)

Ruby used this previously to get 32k data to James so this may be a
solution for you. What I typically do when downloading to Rhodes is
splitting by field (CASA will allow you to transfer calibrate across
databases) - or you can virtual concat

Let me know,

On Mon, Jan 30, 2023 at 10:12 PM IanHeywood ***@***.***>
wrote:

> Trying to use mvftoms.py to pull 32k data from the archive to IDIA. The
> connection frequently drops with:
>
> socket.timeout: The read operation timed out
>
> but when I restart it continues. It's got most of the way there after a
> couple of days (11 TB downloaded), but now I'm getting the error below. I
> get the same error when I try to access the MS using other software. Any
> tips on how to recover this gratefully received. I'll start the transfer
> again in the meantime.
>
> Cheers.
>
> /users/ianh/venv/katdal/lib/python3.8/site-packages/katdal/applycal.py:156: RuntimeWarning: invalid value encountered in reciprocal
>   corrections.append(ComparableArrayWrapper(np.reciprocal(bp)))
> /users/ianh/venv/katdal/lib/python3.8/site-packages/katdal/applycal.py:204: RuntimeWarning: invalid value encountered in reciprocal
>   return np.reciprocal(smooth_gains)
> The following calibration products will be applied: l1.K, l1.B, l1.G, l2.GPHASE
> Per user request the following antennas will be selected: 'm000', 'm001', 'm002', 'm003', 'm004', 'm005', 'm006', 'm007', 'm008', 'm010', 'm012', 'm013', 'm014', 'm015', 'm016', 'm017', 'm018', 'm019', 'm020', 'm021', 'm022', 'm023', 'm024', 'm025', 'm026', 'm027', 'm028', 'm029', 'm030', 'm031', 'm032', 'm033', 'm034', 'm035', 'm036', 'm037', 'm038', 'm039', 'm041', 'm042', 'm043', 'm044', 'm045', 'm046', 'm047', 'm048', 'm049', 'm050', 'm051', 'm052', 'm053', 'm054', 'm055', 'm056', 'm057', 'm058', 'm059', 'm060', 'm061', 'm062', 'm063'
> Per user request the following target fields will be selected: 'COSMOS_3'
> Per user request the following scans will be dumped: 34, 8, 42, 14, 48, 22, 28
> Extract MS for spw 0: centre frequency 1284000000 Hz
> Will create MS output in 1622376680_sdp_l2.full.ms
>
> #### Producing a full polarisation MS (HH,HV,VH,VV) ####
>
>
> Using array as the reference antenna. All targets and scans will be based on this antenna.
>
>
> Iterating through scans in dataset(s)...
>
> Traceback (most recent call last):
>   File "/users/ianh/venv/katdal/bin/mvftoms.py", line 916, in <module>
>     main()
>   File "/users/ianh/venv/katdal/bin/mvftoms.py", line 545, in main
>     with ms_extra.open_table(ms_name, verbose=options.verbose) as t:
>   File "/users/ianh/venv/katdal/lib/python3.8/site-packages/katdal/ms_extra.py", line 42, in open_table
>     return tables.table(name, readonly=readonly, ack=verbose, **kwargs)
>   File "/users/ianh/venv/katdal/lib/python3.8/site-packages/casacore/tables/table.py", line 373, in __init__
>     Table.__init__(self, tabname, lockopt, opt)
> RuntimeError: Table DataManager error: Internal error: TiledStMan::headerFileGet: mismatch in #row; expected 0, found 5313710
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/ska-sa/katdal/issues/362>, or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AB4RE6WZUV7YVL3SIEG4ISDWVAOCZANCNFSM6AAAAAAULSUSUM>
> .
> You are receiving this because you are subscribed to this thread.Message
> ID: ***@***.***>
>


-- 
--
Benjamin Hugo

PhD. student,
Centre for Radio Astronomy Techniques and Technologies
Department of Physics and Electronics
Rhodes University

Junior software developer
Radio Astronomy Research Group
South African Radio Astronomy Observatory
Black River Business Park
Observatory
Cape Town
@ludwigschwardt: When an exception occurs the MS table should have been closed properly but maybe something went wrong there.

I see in the casacore code that there is a TABLEREPAIR option that would allow you to override the row number, but you have to recompile casacore for that (and not sure how well it works at all).

I have an open ticket to address `socket.timeout` issues, but wasn't aware of how often they occur. I'll bump up the priority!@bennahugo: My thoughts on the shape issue is that we are using (or were using) fixed
sized columns. If the correct sizes are not set in the descriptors at the
start it may be part of the problem.

One could switch to variable sized columns but we found early on (back in
the 32-single correlator days) that not using fixed shaped descriptors
leads to an almost an order of magnitude slowdown in the casacore table
system which is not really feasible for reductions

On Tue, Jan 31, 2023 at 9:43 AM Benna Hugo ***@***.***> wrote:

> @IanHeywood - I know this is not a fix but if you look at the help options
> you will see that you can select by scans and fields (rather than
> downloading 11TiB in one go?)
>
> Ruby used this previously to get 32k data to James so this may be a
> solution for you. What I typically do when downloading to Rhodes is
> splitting by field (CASA will allow you to transfer calibrate across
> databases) - or you can virtual concat
>
> Let me know,
>
> On Mon, Jan 30, 2023 at 10:12 PM IanHeywood ***@***.***>
> wrote:
>
>> Trying to use mvftoms.py to pull 32k data from the archive to IDIA. The
>> connection frequently drops with:
>>
>> socket.timeout: The read operation timed out
>>
>> but when I restart it continues. It's got most of the way there after a
>> couple of days (11 TB downloaded), but now I'm getting the error below. I
>> get the same error when I try to access the MS using other software. Any
>> tips on how to recover this gratefully received. I'll start the transfer
>> again in the meantime.
>>
>> Cheers.
>>
>> /users/ianh/venv/katdal/lib/python3.8/site-packages/katdal/applycal.py:156: RuntimeWarning: invalid value encountered in reciprocal
>>   corrections.append(ComparableArrayWrapper(np.reciprocal(bp)))
>> /users/ianh/venv/katdal/lib/python3.8/site-packages/katdal/applycal.py:204: RuntimeWarning: invalid value encountered in reciprocal
>>   return np.reciprocal(smooth_gains)
>> The following calibration products will be applied: l1.K, l1.B, l1.G, l2.GPHASE
>> Per user request the following antennas will be selected: 'm000', 'm001', 'm002', 'm003', 'm004', 'm005', 'm006', 'm007', 'm008', 'm010', 'm012', 'm013', 'm014', 'm015', 'm016', 'm017', 'm018', 'm019', 'm020', 'm021', 'm022', 'm023', 'm024', 'm025', 'm026', 'm027', 'm028', 'm029', 'm030', 'm031', 'm032', 'm033', 'm034', 'm035', 'm036', 'm037', 'm038', 'm039', 'm041', 'm042', 'm043', 'm044', 'm045', 'm046', 'm047', 'm048', 'm049', 'm050', 'm051', 'm052', 'm053', 'm054', 'm055', 'm056', 'm057', 'm058', 'm059', 'm060', 'm061', 'm062', 'm063'
>> Per user request the following target fields will be selected: 'COSMOS_3'
>> Per user request the following scans will be dumped: 34, 8, 42, 14, 48, 22, 28
>> Extract MS for spw 0: centre frequency 1284000000 Hz
>> Will create MS output in 1622376680_sdp_l2.full.ms
>>
>> #### Producing a full polarisation MS (HH,HV,VH,VV) ####
>>
>>
>> Using array as the reference antenna. All targets and scans will be based on this antenna.
>>
>>
>> Iterating through scans in dataset(s)...
>>
>> Traceback (most recent call last):
>>   File "/users/ianh/venv/katdal/bin/mvftoms.py", line 916, in <module>
>>     main()
>>   File "/users/ianh/venv/katdal/bin/mvftoms.py", line 545, in main
>>     with ms_extra.open_table(ms_name, verbose=options.verbose) as t:
>>   File "/users/ianh/venv/katdal/lib/python3.8/site-packages/katdal/ms_extra.py", line 42, in open_table
>>     return tables.table(name, readonly=readonly, ack=verbose, **kwargs)
>>   File "/users/ianh/venv/katdal/lib/python3.8/site-packages/casacore/tables/table.py", line 373, in __init__
>>     Table.__init__(self, tabname, lockopt, opt)
>> RuntimeError: Table DataManager error: Internal error: TiledStMan::headerFileGet: mismatch in #row; expected 0, found 5313710
>>
>> —
>> Reply to this email directly, view it on GitHub
>> <https://github.com/ska-sa/katdal/issues/362>, or unsubscribe
>> <https://github.com/notifications/unsubscribe-auth/AB4RE6WZUV7YVL3SIEG4ISDWVAOCZANCNFSM6AAAAAAULSUSUM>
>> .
>> You are receiving this because you are subscribed to this thread.Message
>> ID: ***@***.***>
>>
>
>
> --
> --
> Benjamin Hugo
>
> PhD. student,
> Centre for Radio Astronomy Techniques and Technologies
> Department of Physics and Electronics
> Rhodes University
>
> Junior software developer
> Radio Astronomy Research Group
> South African Radio Astronomy Observatory
> Black River Business Park
> Observatory
> Cape Town
>


-- 
--
Benjamin Hugo

PhD. student,
Centre for Radio Astronomy Techniques and Technologies
Department of Physics and Electronics
Rhodes University

Junior software developer
Radio Astronomy Research Group
South African Radio Astronomy Observatory
Black River Business Park
Observatory
Cape Town
@ludwigschwardt: Thanks @bennahugo. The `mvftoms` script typically opens the table and then adds some rows, one dump at a time. It can now resume, which means that it checks the number of rows in the existing MS and continues from there. Somewhere in the process the row counter was trashed, if you ask me.@IanHeywood: > I know this is not a fix but if you look at the help options you will see that you can select by scans and fields (rather than downloading 11TiB in one go?)

Thanks @bennahugo, @ludwigschwardt. 

I'm already selecting only the target field since it's transferring L2 data. Splitting by scan might be an option, although I'll have to modify the processing scripts and up my bookkeeping game. Having the twin curse of being both busy and lazy I'd rather avoid it.

On that note I've just removed the broken MS and restarted the transfer. It's already almost half way.

Cheers.@IanHeywood: > I have an open ticket to address socket.timeout issues, but wasn't aware of how often they occur. I'll bump up the priority!

I did search for this prior to starting a new issue, but couldn't find it. Can you please link it here if you get a moment?

Thanks!@ludwigschwardt: It's an internal ticket: https://skaafrica.atlassian.net/browse/SPR1-813 (but maybe you could still access it, since I see you have a Jira account).@IanHeywood: I don't have permission, but don't worry about it. Was just going to follow the developments. Cheers.@Jordatious: @ludwigschwardt, this is the same issue we discussed in November. My error was much the same:

> RuntimeError: Table DataManager error: Internal error: TiledStMan::headerFileGet: mismatch in #row; expected 735660, found 739320

For me, the issue arose from the SLURM job timing out, so the script(s) didn't have a chance to close the table gracefully. From memory, your suggested solution was to update the metadata so that it either expects the higher number of rows (assuming they're intact), or remove the newer rows beyond what's expected, so that they can be written again. How possible are either of those solutions?

> I was looking to implement a (perhaps slightly dodgy) script that captures the mvftoms.py output and checks the last line for the time out error and then restarts the transfer if it's there.

I've given @IanHeywood a copy of my method for doing this with SLURM, which uses the requeue option. If it's generally useful for other users, I can post it somewhere. What do you think?@IanHeywood: Just a note that the timeout issue seems to be an actual showstopper right now. Whereas previously it might do a few scans, timeout and then resume, for the last 1-2 days it's been timing out before it manages to complete the transfer of a single scan, thus restarting does not advance things.@Jordatious: I'm also having trouble, although I seem to have gotten further (dump 520, 77, 225). I'm on my 3rd attempt to transfer the same MS over the past few days, which initially wrote 1.1 TB before continually failing, 188 GB, and just now, 551 GB.@ludwigschwardt: I'm very sorry to hear that! I'll see if I can cook up a solution in the next few days.@IanHeywood: Getting somewhat off-topic now, but just wanted a record of this new flavour of download interruption:

```
  File "/users/ianh/venv/katdal/lib/python3.8/site-packages/katsdptelstate/telescope_state.py", line 172, in load_from_file
    keys_loaded = self._backend.load_from_file(file)
  File "/users/ianh/venv/katdal/lib/python3.8/site-packages/katsdptelstate/memory.py", line 177, in load_from_file
    return rdb_reader.load_from_file(MemoryCallback(self._data), file)
  File "/users/ianh/venv/katdal/lib/python3.8/site-packages/katsdptelstate/rdb_reader.py", line 84, in load_from_file
    _parse_rdb_file(parser, callback, cast(BinaryIO, file))
  File "/users/ianh/venv/katdal/lib/python3.8/site-packages/katsdptelstate/rdb_reader.py", line 50, in _parse_rdb_file
    raise RdbParseError(filename) from exc
katsdptelstate.errors.RdbParseError: Invalid RDB file object
```

It looks like the RDB token has spontaneously changed. Don't know if that's expected behaviour!

Cheers.@ludwigschwardt: I think what's happening is that the RDB file itself is truncated during download and therefore could not open. The token should still be valid.

I currently do not retry the RDB download... It's also on the to-do list as a separate but related issue.
@ludwigschwardt: Reopening this while it is being fixed.@Jordatious: Hi @ludwigschwardt, I had a go of a remote `mvftoms.py` pull from ilifu using [this branch](https://github.com/ska-sa/katdal/tree/spr1-813-handle-read-errors) of katdal last month after I noticed your pull request recently, and found it worked quite well, even for large pulls (5.7 TiB and 2.4 TiB, requeued 89 and 56 times, respectively). 

I've just started to use it again now, and immediately had the following error thrown upon trying to `import numba`. I'm wondering if something has changed that might cause this error, either on your side or ours (e.g. we recently migrated a number of our VMs to bare metal machines). I started using a more recent stable version of katdal (0.20.2.dev1621+master.eb8db22, using numba-0.56.4 and numpy-1.23.4) and it started pulling immediately. I've also tried updating numba for my venv of katdal using the dev branch, which was unchanged as version 0.56.4, but rolled back to numpy-1.23.5 (from 1.24.2) and installed style-1.1.0 and update-0.0.1, and that also immediately started pulling. If you have any clarity on the issue, please let me know!

```
Traceback (most recent call last):
  File "/idia/users/jcollier/software/katdal-dev/bin/mvftoms.py", line 35, in <module>
    import numba
  File "/idia/users/jcollier/software/katdal-dev/lib/python3.8/site-packages/numba/__init__.py", line 42, in <module>
    from numba.np.ufunc import (vectorize, guvectorize, threading_layer,
  File "/idia/users/jcollier/software/katdal-dev/lib/python3.8/site-packages/numba/np/ufunc/__init__.py", line 3, in <module>
    from numba.np.ufunc.decorators import Vectorize, GUVectorize, vectorize, guvectorize
  File "/idia/users/jcollier/software/katdal-dev/lib/python3.8/site-packages/numba/np/ufunc/decorators.py", line 3, in <module>
    from numba.np.ufunc import _internal
SystemError: initialization of _internal failed without raising an exception
```@bmerry: Sounds like https://github.com/numba/numba/issues/8615@Jordatious: Ok thanks @bmerry. I ended up using the stable version I have of katdal and two ~1.6 TiB MS transfers completed after only ~4 times of having to requeue!@Jordatious: Hi all, just a quick update on this. This week I pulled 3 MSs to ilifu using `mvftoms.py` from the #363 [dev branch](https://github.com/ska-sa/katdal/tree/spr1-813-handle-read-errors) (with numpy-1.23.5), and all 3 completed in one attempt! They respectively took roughly 10.5, 11.5 and 20 hours to pull roughly 1, 1.1 and 2.6 TiB.@ludwigschwardt: Excellent news! I think it's time to merge the PR.@ludwigschwardt: This was fixed by #363.

Solution patch:
diff --git a/katdal/chunkstore_s3.py b/katdal/chunkstore_s3.py
index adf6861..a99ca98 100644
--- a/katdal/chunkstore_s3.py
+++ b/katdal/chunkstore_s3.py
@@ -20,6 +20,7 @@ import base64
 import contextlib
 import copy
 import hashlib
+import http.client
 import json
 import threading
 import time
@@ -30,14 +31,19 @@ import urllib.request
 import jwt
 import numpy as np
 import requests
+import urllib3
 
 try:
     import botocore.auth
     import botocore.credentials
 except ImportError:
     botocore = None
-from urllib3.exceptions import MaxRetryError
-from urllib3.response import HTTPResponse
+try:
+    # XXX A deprecated alias of TimeoutError since Python 3.10
+    from socket import timeout as SocketTimeoutError
+except ImportError:
+    SocketTimeoutError = TimeoutError
+from urllib3.exceptions import MaxRetryError, ReadTimeoutError, ProtocolError, IncompleteRead
 from urllib3.util.retry import Retry
 
 from .chunkstore import (BadChunk, ChunkNotFound, ChunkStore, StoreUnavailable,
@@ -68,11 +74,9 @@ _BUCKET_POLICY = {
     ]
 }
 
-# Fake HTTP status code associated with reset connections / truncated data
-_TRUNCATED_HTTP_STATUS_CODE = 555
 # These HTTP responses typically indicate temporary S3 server / proxy overload,
 # which will trigger retries terminating in a missing data response if unsuccessful.
-_DEFAULT_SERVER_GLITCHES = (500, 502, 503, 504, _TRUNCATED_HTTP_STATUS_CODE)
+_DEFAULT_SERVER_GLITCHES = (500, 502, 503, 504)
 
 
 class S3ObjectNotFound(ChunkNotFound):
@@ -82,33 +86,50 @@ class S3ObjectNotFound(ChunkNotFound):
 class S3ServerGlitch(ChunkNotFound):
     """S3 chunk store responded with an HTTP error deemed to be temporary."""
 
-    def __init__(self, msg, status_code):
-        super().__init__(msg)
-        self.status_code = status_code
-
-
-class TruncatedRead(ValueError):
-    """HTTP request to S3 chunk store responded with fewer bytes than expected."""
-
 
 class _DetectTruncation:
-    """Raise :exc:`TruncatedRead` if wrapped `readable` runs out of data."""
+    """Raise :exc:`IncompleteRead` if wrapped `readable` runs out of data."""
 
     def __init__(self, readable):
         self._readable = readable
+        if isinstance(readable, http.client.HTTPResponse):
+            # Store initial length in case of httplib so that we can figure out bytes_read
+            # (because tell() doesn't work since the socket cannot be seeked).
+            self._content_length = readable.length
+        else:
+            self._content_length = None
 
     def __getattr__(self, name):
         """Proxy all attributes to underlying wrapped object."""
         return getattr(self._readable, name)
 
-    def read(self, size, *args, **kwargs):
+    def _raise_incomplete_read(self, bytes_read, bytes_left):
+        """Figure out more accurate parameters for `IncompleteRead` and raise it."""
+        if isinstance(self._readable, http.client.HTTPResponse):
+            if self._readable.length is not None:
+                bytes_left = self._readable.length
+                bytes_read = self._content_length - bytes_left
+        elif isinstance(self._readable, urllib3.response.HTTPResponse):
+            if self._readable.length_remaining is not None:
+                bytes_left = self._readable.length_remaining
+                bytes_read = self._readable.tell()
+        raise IncompleteRead(bytes_read, bytes_left)
+
+    def read(self, size=None, *args, **kwargs):
         """Overload `read` method to detect truncated data source."""
         data = self._readable.read(size, *args, **kwargs)
-        if data == b'' and size != 0:
-            raise TruncatedRead('Error reading from S3 HTTP response: expected '
-                                f'{size} more byte(s), got EOF')
+        if data == b'' and size is not None and size > 0:
+            self._raise_incomplete_read(0, size)
         return data
 
+    def readinto(self, buffer, *args, **kwargs):
+        """Overload `readinto` method to detect truncated data source."""
+        view = memoryview(buffer)
+        bytes_read = self._readable.readinto(view, *args, **kwargs)
+        if bytes_read != view.nbytes:
+            self._raise_incomplete_read(bytes_read, view.nbytes - bytes_read)
+        return bytes_read
+
 
 def read_array(fp):
     """Read a numpy array in npy format from a file descriptor.
@@ -138,10 +159,7 @@ def read_array(fp):
     # For HTTPResponse it works to just pass in `data` directly, but the
     # wrapping is added for the benefit of any other implementation that
     # isn't expecting a numpy array
-    bytes_read = fp.readinto(memoryview(data.view(np.uint8)))
-    if bytes_read != data.nbytes:
-        raise TruncatedRead('Error reading from S3 HTTP response: '
-                            f'expected {data.nbytes} bytes, got {bytes_read}')
+    fp.readinto(data.view(np.uint8))
     if fortran_order:
         data.shape = shape[::-1]
         data = data.transpose()
@@ -168,6 +186,16 @@ def _read_chunk(response):
     return chunk
 
 
+def _read_object(response):
+    """Read bytes from content of HTTP response and check that it's all there."""
+    data = response.content
+    # Verify that all content has been consumed
+    bytes_left = response.raw.length_remaining
+    if bytes_left is not None and bytes_left > 0:
+        raise IncompleteRead(response.raw.tell(), bytes_left)
+    return data
+
+
 def _bucket_url(url):
     """Turn `url` into associated S3 bucket URL (first path component only)."""
     split_url = urllib.parse.urlsplit(url)
@@ -396,6 +424,66 @@ class _Multipart:
         return sum(memoryview(item).nbytes for item in self.items)
 
 
+def _raise_for_status(response, chunk_name, ignored_errors):
+    """Turn error responses into appropriate exceptions, like raise_for_status."""
+    status = response.status_code
+    if 400 <= status < 600 and status not in ignored_errors:
+        # Construct error message, including detailed response content if sensible
+        prefix = f'Chunk {chunk_name!r}: ' if chunk_name else ''
+        msg = (f'{prefix}Store responded with HTTP error {status} ({response.reason}) '
+               f'to request: {response.request.method} {response.url}')
+        content_type = response.headers.get('Content-Type')
+        if content_type in ('application/xml', 'text/xml', 'text/plain'):
+            msg += f'\nDetails of server response: {response.text}'
+        # Raise the appropriate exception
+        if status in (401, 403):
+            raise AuthorisationFailed(msg)
+        elif status == 404:
+            raise S3ObjectNotFound(msg)
+        else:
+            raise StoreUnavailable(msg)
+
+
+@contextlib.contextmanager
+def _request(session, method, url, timeout=(None, None), **kwargs):
+    """A beefed-up request that facilitates retries on reading the response.
+
+    This catches socket timeouts and reset connections while the response data
+    is being read and reraises them as appropriate urllib3 exceptions that can
+    be passed to a `Retry` object to trigger a read retry.
+    """
+    try:
+        with session.request(method, url, timeout=timeout, **kwargs) as response:
+            yield response
+    except SocketTimeoutError as error:
+        msg = f'Read timed out - socket idle for {timeout[1]} seconds'
+        raise ReadTimeoutError('Read timeout', url, msg) from error
+    except (
+        # Requests massages ProtocolErrors into ChunkedEncodingErrors, so turn it back
+        requests.exceptions.ChunkedEncodingError,
+        ConnectionResetError,
+        IncompleteRead,
+    ) as error:
+        raise ProtocolError(str(error)) from error
+
+
+def _connect_read_tuple(connect_and_or_read):
+    """Turn connect and/or read retries/timeouts into a (connect, read) tuple."""
+    try:
+        connect, read = connect_and_or_read
+    except TypeError:
+        connect = read = connect_and_or_read
+    return (connect, read)
+
+
+def _retry_object(retries, **defaults):
+    """Turn (connect, read) `retries` into `Retry` object (or keep as is)."""
+    if not isinstance(retries, Retry):
+        connect_retries, read_retries = _connect_read_tuple(retries)
+        retries = Retry(connect=connect_retries, read=read_retries, **defaults)
+    return retries
+
+
 class S3ChunkStore(ChunkStore):
     """A store of chunks (i.e. N-dimensional arrays) based on the Amazon S3 API.
 
@@ -421,9 +509,9 @@ class S3ChunkStore(ChunkStore):
         string type with UTF-8. The URL may also contain a path if this store is
         relative to an existing bucket, in which case the chunk name is a relative
         path (useful for unit tests).
-    timeout : float or tuple of 2 floats, optional
+    timeout : float or None or tuple of 2 floats or None's, optional
         Connect / read timeout, in seconds, either a single value for both or
-        custom values as (connect, read) tuple (set to None to leave unchanged)
+        custom values as (connect, read) tuple. None means "wait forever"...
     retries : int or tuple of 2 ints or :class:`urllib3.util.retry.Retry`, optional
         Number of connect / read retries, either a single value for both or
         custom values as (connect, read) tuple, or a `Retry` object for full
@@ -449,35 +537,34 @@ class S3ChunkStore(ChunkStore):
 
     def __init__(self, url, timeout=(30, 300), retries=2, token=None,
                  credentials=None, public_read=False, expiry_days=0, **kwargs):
-        error_map = {requests.exceptions.RequestException: StoreUnavailable}
+        error_map = {
+            # Urllib3 / Requests exceptions raised when read / status retries run out
+            MaxRetryError: S3ServerGlitch,  # too many retries on reading response
+            requests.exceptions.ReadTimeout: S3ServerGlitch,  # read timeouts on header
+            requests.exceptions.RetryError: S3ServerGlitch,  # too many status retries
+            # A generic request error (includes connection failures)
+            requests.exceptions.RequestException: StoreUnavailable,
+        }
         super().__init__(error_map)
         auth = _auth_factory(url, token, credentials)
-        if not isinstance(retries, Retry):
-            try:
-                connect_retries, read_retries = retries
-            except TypeError:
-                connect_retries = read_retries = retries
-            # The backoff factor of 10 provides 5 minutes worth of retries
-            # when the S3 server is strained; with 5 retries you get
-            # (0 + 2 + 4 + 8 + 16) * 10 = 300 seconds on top of read timeouts.
-            retries = Retry(connect=connect_retries, read=read_retries,
-                            status=5, backoff_factor=10.,
-                            status_forcelist=_DEFAULT_SERVER_GLITCHES)
 
         def session_factory():
             session = _CacheSettingsSession(url)
             session.auth = auth
-            # Don't let requests do status retries as we'll be doing it ourselves
-            max_retries = retries.new(status=0, raise_on_status=False)
-            adapter = requests.adapters.HTTPAdapter(max_retries=max_retries)
+            # Don't set `max_retries` yet - it will be done at the request level
+            adapter = requests.adapters.HTTPAdapter()
             session.mount(url, adapter)
             return session
 
         self._session_pool = _Pool(session_factory)
         self._url = to_str(url)
-        self._retries = retries
         self._verified_buckets = set()
-        self.timeout = timeout
+        self.timeout = _connect_read_tuple(timeout)
+        # The backoff factor of 10 provides 5 minutes worth of retries
+        # when the S3 server is strained; with 5 retries you get
+        # (0 + 2 + 4 + 8 + 16) * 10 = 300 seconds on top of read timeouts.
+        self.retries = _retry_object(retries, status=5, backoff_factor=10.,
+                                     status_forcelist=_DEFAULT_SERVER_GLITCHES)
         self.public_read = public_read
         self.expiry_days = int(expiry_days)
 
@@ -485,93 +572,39 @@ class S3ChunkStore(ChunkStore):
         """Assemble URL corresponding to chunk (or array) name."""
         return urllib.parse.urljoin(self._url, to_str(urllib.parse.quote(chunk_name + extension)))
 
-    @contextlib.contextmanager
-    def request(self, method, url, chunk_name='', ignored_errors=(), timeout=(), **kwargs):
-        """Run a request on a session from the pool and handle error responses.
+    def request(
+        self,
+        method,
+        url,
+        process=lambda response: response,
+        chunk_name='',
+        ignored_errors=(),
+        timeout=(),  # None has a special meaning, so use something else to indicate default
+        retries=None,
+        **kwargs
+    ):
+        """Send HTTP request to S3 server, process response and retry if needed.
 
-        This is a context manager like :meth:`requests.Session.request` that
-        either yields a successful response or raises the appropriate chunk
-        store exception.
+        This retries temporary HTTP errors, including reset connections while
+        processing a successful response.
 
         Parameters
         ----------
         method, url : str
             The standard required parameters of :meth:`requests.Session.request`
+        process : function, signature ``result = process(response)``, optional
+            Function that will process response (just return response by default)
         chunk_name : str, optional
             Name of chunk, used for error reporting only
         ignored_errors : collection of int, optional
             HTTP status codes that are treated like 200 OK, not raising an error
-        timeout : tuple or float, optional
+        timeout : float or None or tuple of 2 floats or None's, optional
             Override timeout for this request (use the store timeout by default)
+        retries : int or tuple of 2 ints or :class:`urllib3.util.retry.Retry`, optional
+            Override retries for this request (use the store retries by default)
         kwargs : optional
             These are passed on to :meth:`requests.Session.request`
 
-        Yields
-        ------
-        response : `requests.Response`
-            HTTP response, considered successful
-
-        Raises
-        ------
-        AuthorisationFailed
-            If the request is not authorised by appropriate token or credentials
-        S3ObjectNotFound
-            If S3 object request fails because it does not exist
-        S3ServerGlitch
-            If HTTP error has a retryable status code (i.e. temporary glitch)
-        StoreUnavailable
-            If a general HTTP error occurred that is not ignored
-        """
-        kwargs['timeout'] = self.timeout if timeout == () else timeout
-        # Use _standard_errors to filter errors emanating from within with-block
-        with self._standard_errors(chunk_name), self._session_pool() as session:
-            with session.request(method, url, **kwargs) as response:
-                # Turn error responses into the appropriate exception, like raise_for_status
-                status = response.status_code
-                if 400 <= status < 600 and status not in ignored_errors:
-                    # Construct error message, including detailed response content if sensible
-                    prefix = f'Chunk {chunk_name!r}: ' if chunk_name else ''
-                    msg = (f'{prefix}Store responded with HTTP error {status} ({response.reason}) '
-                           f'to request: {response.request.method} {response.url}')
-                    content_type = response.headers.get('Content-Type')
-                    if content_type in ('application/xml', 'text/xml', 'text/plain'):
-                        msg += f'\nDetails of server response: {response.text}'
-                    # Raise the appropriate exception
-                    if status in (401, 403):
-                        raise AuthorisationFailed(msg)
-                    elif status == 404:
-                        raise S3ObjectNotFound(msg)
-                    elif self._retries.is_retry(method, status):
-                        raise S3ServerGlitch(msg, status)
-                    else:
-                        raise StoreUnavailable(msg)
-                try:
-                    yield response
-                except TruncatedRead as trunc_error:
-                    # A truncated read is considered a glitch with custom status
-                    prefix = f'Chunk {chunk_name!r}: ' if chunk_name else ''
-                    glitch_error = S3ServerGlitch(prefix + str(trunc_error),
-                                                  _TRUNCATED_HTTP_STATUS_CODE)
-                    raise glitch_error from trunc_error
-
-    def complete_request(self, method, url, chunk_name='',
-                         process=lambda response: None, **kwargs):
-        """Send HTTP request to S3 server, process response and retry if needed.
-
-        This retries temporary HTTP errors, including reset connections while
-        processing a successful response.
-
-        Parameters
-        ----------
-        method, url : str
-            The standard required parameters of :meth:`requests.Session.request`
-        chunk_name : str, optional
-            Name of chunk, used for error reporting only
-        process : function, signature ``result = process(response)``, optional
-            Function that will process response (the default does nothing)
-        kwargs : optional
-            Passed on to :meth:`request` and :meth:`requests.Session.request`
-
         Returns
         -------
         result : object
@@ -588,23 +621,24 @@ class S3ChunkStore(ChunkStore):
         StoreUnavailable
             If a general HTTP error occurred that is not ignored
         """
-        retries = self._retries.new()
-        while True:
-            try:
-                with self.request(method, url, chunk_name, **kwargs) as response:
-                    result = process(response)
-            except S3ServerGlitch as e:
-                # Retry based on status of response until we run out of retries
-                response = HTTPResponse(status=e.status_code)
+        timeout = self.timeout if timeout == () else _connect_read_tuple(timeout)
+        retries = self.retries if retries is None else _retry_object(retries)
+        retries = retries.new()
+        # Use _standard_errors to filter errors emanating from within with-block
+        with self._standard_errors(chunk_name), self._session_pool() as session:
+            adapter = session.get_adapter(url)
+            while True:
+                # Initialise and reuse the same Retry object for the entire session
+                adapter.max_retries = retries
                 try:
-                    retries = retries.increment(method, url, response)
-                except MaxRetryError:
-                    # Raise the final straw that broke the retry camel's back
-                    raise e from None
-                else:
-                    retries.sleep(response)
-            else:
-                return result
+                    with _request(session, method, url, timeout, **kwargs) as response:
+                        _raise_for_status(response, chunk_name, ignored_errors)
+                        retries = response.raw.retries.new()
+                        return process(response)
+                # Urllib3 exceptions that can trigger read retries
+                except (ReadTimeoutError, ProtocolError) as error:
+                    retries = retries.increment(method, url, error=error)
+                    retries.sleep()
 
     def _verify_bucket(self, url, chunk_error=None):
         """Check that bucket associated with `url` exists and is not empty."""
@@ -613,8 +647,7 @@ class S3ChunkStore(ChunkStore):
             return
         try:
             # Speed up the request by only checking that the bucket has at least one key
-            response = self.complete_request('GET', bucket, process=lambda r: r,
-                                             params={'max-keys': 1})
+            response = self.request('GET', bucket, params={'max-keys': 1})
         except S3ObjectNotFound as err:
             # There is no point continuing if the bucket is completely missing
             raise StoreUnavailable(err) from chunk_error
@@ -634,8 +667,8 @@ class S3ChunkStore(ChunkStore):
         # work with non-identity encodings.
         headers = {'Accept-Encoding': 'identity'}
         try:
-            chunk = self.complete_request('GET', url, chunk_name, _read_chunk,
-                                          headers=headers, stream=True)
+            chunk = self.request('GET', url, _read_chunk, chunk_name=chunk_name,
+                                 headers=headers, stream=True)
         except S3ObjectNotFound as err:
             # If the entire bucket is gone, this becomes StoreUnavailable instead
             self._verify_bucket(url, err)
@@ -655,7 +688,7 @@ class S3ChunkStore(ChunkStore):
 
     def _create_bucket(self, url):
         # Make bucket (409 indicates the bucket already exists, which is OK)
-        self.complete_request('PUT', url, ignored_errors=(409,))
+        self.request('PUT', url, ignored_errors=(409,))
 
         if self.public_read:
             policy_url = urllib.parse.urljoin(url, '?policy')
@@ -665,14 +698,14 @@ class S3ChunkStore(ChunkStore):
                 f'arn:aws:s3:::{bucket_name}/*',
                 f'arn:aws:s3:::{bucket_name}'
             ]
-            self.complete_request('PUT', policy_url, data=json.dumps(policy))
+            self.request('PUT', policy_url, data=json.dumps(policy))
 
         if self.expiry_days > 0:
             xml_payload = _BASE_LIFECYCLE_POLICY.format(self.expiry_days)
             b64_md5 = base64.b64encode(hashlib.md5(xml_payload.encode('utf-8')).digest()).decode('utf-8')
             lifecycle_headers = {'Content-Type': 'text/xml', 'Content-MD5': b64_md5}
-            self.complete_request('PUT', url, params='lifecycle',
-                                  data=xml_payload, headers=lifecycle_headers)
+            self.request('PUT', url, params='lifecycle',
+                         data=xml_payload, headers=lifecycle_headers)
 
     def put_chunk(self, array_name, slices, chunk):
         """See the docstring of :meth:`ChunkStore.put_chunk`."""
@@ -686,21 +719,21 @@ class S3ChunkStore(ChunkStore):
         md5 = base64.b64encode(md5_gen.digest())
         headers = {'Content-MD5': md5.decode()}
         data = _Multipart([npy_header, memoryview(chunk)])
-        self.complete_request('PUT', url, chunk_name, headers=headers, data=data)
+        self.request('PUT', url, chunk_name=chunk_name, headers=headers, data=data)
 
     def mark_complete(self, array_name):
         """See the docstring of :meth:`ChunkStore.mark_complete`."""
         self.create_array(array_name)
         obj_name = self.join(array_name, 'complete')
         url = urllib.parse.urljoin(self._url, obj_name)
-        self.complete_request('PUT', url, obj_name, data=b'')
+        self.request('PUT', url, chunk_name=obj_name, data=b'')
 
     def is_complete(self, array_name):
         """See the docstring of :meth:`ChunkStore.is_complete`."""
         obj_name = self.join(array_name, 'complete')
         url = urllib.parse.urljoin(self._url, obj_name)
         try:
-            self.complete_request('GET', url, obj_name)
+            self.request('GET', url, chunk_name=obj_name)
         except ChunkNotFound:
             return False
         return True
diff --git a/katdal/datasources.py b/katdal/datasources.py
index 0c7fc94..9fa9f54 100644
--- a/katdal/datasources.py
+++ b/katdal/datasources.py
@@ -26,7 +26,7 @@ import numpy as np
 
 from .chunkstore import ChunkStoreError
 from .chunkstore_npy import NpyFileChunkStore
-from .chunkstore_s3 import S3ChunkStore
+from .chunkstore_s3 import S3ChunkStore, _read_object
 from .dataset import parse_url_or_path
 from .sensordata import TelstateSensorGetter, TelstateToStr
 from .vis_flags_weights import ChunkStoreVisFlagsWeights
@@ -461,8 +461,8 @@ class TelstateDataSource(DataSource):
             telstate = katsdptelstate.TelescopeState()
             try:
                 rdb_store = S3ChunkStore(store_url, **kwargs)
-                with rdb_store.request('GET', rdb_url) as response:
-                    telstate.load_from_file(io.BytesIO(response.content))
+                rdb_data = rdb_store.request('GET', rdb_url, process=_read_object)
+                telstate.load_from_file(io.BytesIO(rdb_data))
             except ChunkStoreError as e:
                 raise DataSourceNotFound(str(e)) from e
             # If the RDB file is opened via archive URL, use that URL and


Test patch:
diff --git a/katdal/test/test_chunkstore_s3.py b/katdal/test/test_chunkstore_s3.py
index 0897da6..44687d7 100644
--- a/katdal/test/test_chunkstore_s3.py
+++ b/katdal/test/test_chunkstore_s3.py
@@ -35,6 +35,7 @@ import pathlib
 import re
 import shutil
 import socket
+import struct
 import sys
 import tempfile
 import threading
@@ -49,13 +50,22 @@ import requests
 from katsdptelstate.rdb_writer import RDBWriter
 import pytest
 from numpy.testing import assert_array_equal
+from urllib3.exceptions import IncompleteRead
 from urllib3.util.retry import Retry
 
 from katdal.chunkstore import ChunkNotFound, StoreUnavailable
-from katdal.chunkstore_s3 import (_DEFAULT_SERVER_GLITCHES, InvalidToken,
-                                  S3ChunkStore, TruncatedRead, _AWSAuth,
-                                  AuthorisationFailed, decode_jwt, read_array)
-from katdal.datasources import TelstateDataSource
+from katdal.chunkstore_s3 import (
+    _DEFAULT_SERVER_GLITCHES,
+    InvalidToken,
+    S3ChunkStore,
+    S3ServerGlitch,
+    _AWSAuth,
+    AuthorisationFailed,
+    decode_jwt,
+    read_array,
+    _read_object,
+)
+from katdal.datasources import TelstateDataSource, DataSourceNotFound
 from katdal.test.s3_utils import MissingProgram, S3Server, S3User
 from katdal.test.test_chunkstore import ChunkStoreTestBase, generate_arrays
 from katdal.test.test_datasources import (assert_telstate_data_source_equal,
@@ -68,11 +78,11 @@ PREFIX = '1234567890'
 # Pick quick but different timeouts and retries for unit tests:
 #  - The effective connect timeout is 5.0 (initial) + 5.0 (1 retry) = 10 seconds
 #  - The effective read timeout is 2.0 + 3 * 2.0 + 0.1 * (0 + 2 + 4) = 8.6 seconds
-#  - The effective status timeout is 0.1 * (0 + 2 + 4) = 0.6 seconds, or
-#    4 * 0.1 + 0.6 = 1.0 second if the suggestions use SUGGESTED_STATUS_DELAY
+#  - The effective status timeout is 0.1 * (0 + 2) = 0.2 seconds, or
+#    3 * 0.1 + 0.2 = 0.5 second if the suggestions use SUGGESTED_STATUS_DELAY
 TIMEOUT = (5.0, 2.0)
-RETRY = Retry(connect=1, read=3, status=3, backoff_factor=0.1,
-              raise_on_status=False, status_forcelist=_DEFAULT_SERVER_GLITCHES)
+RETRY = Retry(connect=1, read=3, status=2, backoff_factor=0.1,
+              status_forcelist=_DEFAULT_SERVER_GLITCHES)
 SUGGESTED_STATUS_DELAY = 0.1
 READ_PAUSE = 0.1
 # Dummy private key for ES256 algorithm (taken from PyJWT unit tests)
@@ -103,9 +113,9 @@ def get_free_port(host):
 
 
 class TestReadArray:
-    def _test(self, array):
+    def _test(self, array, version=None):
         fp = io.BytesIO()
-        np.save(fp, array)
+        np.lib.format.write_array(fp, array, version, allow_pickle=False)
         fp.seek(0)
         out = read_array(fp)
         np.testing.assert_equal(array, out)
@@ -122,12 +132,7 @@ class TestReadArray:
         self._test(np.arange(20).reshape(4, 5, 1).T)
 
     def testV2(self):
-        # Make dtype that needs more than 64K to store, forcing .npy version 2.0
-        dtype = np.dtype([('a' * 70000, np.float32), ('b', np.float32)])
-        with warnings.catch_warnings():
-            # Suppress warning that V2 files can only be read by numpy >= 1.9
-            warnings.simplefilter('ignore', category=UserWarning)
-            self._test(np.zeros(100, dtype))
+        self._test(np.zeros(100), version=(2, 0))
 
     def testBadVersion(self):
         data = b'\x93NUMPY\x03\x04'     # Version 3.4
@@ -149,7 +154,7 @@ class TestReadArray:
         fp.seek(*args)
         fp.truncate()
         fp.seek(0)
-        with pytest.raises(TruncatedRead):
+        with pytest.raises(IncompleteRead):
             read_array(fp)
 
     def testShort(self):
@@ -273,6 +278,8 @@ class TestS3ChunkStore(ChunkStoreTestBase):
                 shutil.rmtree(entry.path)
         # Also get rid of the cache of verified buckets
         self.store._verified_buckets.clear()
+        self.store.timeout = TIMEOUT
+        self.store.retries = RETRY
         print(f"Chunk store: {self.store_url}, S3 server: {self.s3_url}")
 
     def array_name(self, name):
@@ -326,7 +333,7 @@ class TestS3ChunkStore(ChunkStoreTestBase):
     def test_mark_complete_top_level(self):
         self._test_mark_complete(PREFIX + '-completetest')
 
-    def test_rdb_support(self):
+    def test_rdb_support(self, suggestion=''):
         telstate = katsdptelstate.TelescopeState()
         view, cbid, sn, _, _ = make_fake_data_source(telstate, self.store, (5, 16, 40), PREFIX)
         telstate['capture_block_id'] = cbid
@@ -339,9 +346,13 @@ class TestS3ChunkStore(ChunkStoreTestBase):
         # Read the file back in and upload it to S3
         with open(temp_filename, mode='rb') as rdb_file:
             rdb_data = rdb_file.read()
-        rdb_url = urllib.parse.urljoin(self.store_url, self.store.join(cbid, rdb_filename))
+        if suggestion:
+            rdb_path = self.store.join(suggestion, cbid, rdb_filename)
+        else:
+            rdb_path = self.store.join(cbid, rdb_filename)
+        rdb_url = urllib.parse.urljoin(self.store_url, rdb_path)
         self.store.create_array(cbid)
-        self.store.complete_request('PUT', rdb_url, data=rdb_data)
+        self.store.request('PUT', rdb_url, data=rdb_data)
         # Check that data source can be constructed from URL (with auto chunk store)
         source_from_url = TelstateDataSource.from_url(rdb_url, **self.store_kwargs)
         source_direct = TelstateDataSource(view, cbid, sn, self.store)
@@ -374,6 +385,17 @@ class _TokenHTTPProxyHandler(http.server.BaseHTTPRequestHandler):
             return self.do_all
         return self.__getattribute__(name)
 
+    def _reset_connection(self):
+        """Send a TCP reset (RST) packet to reset the connection.
+
+        Enable SO_LINGER with a linger interval of 0 s. This drops the
+        connection like a hot potato once closed.
+        """
+        l_onoff = 1  # non-zero value enables linger option in kernel
+        l_linger = 0  # timeout interval in seconds
+        self.connection.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
+                                   struct.pack('ii', l_onoff, l_linger))
+
     def do_all(self):
         # See https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Connection
         HOP_HEADERS = {
@@ -385,6 +407,7 @@ class _TokenHTTPProxyHandler(http.server.BaseHTTPRequestHandler):
         data = self.rfile.read(data_len)
         truncate = False
         pause = 0.0
+        reset = False
         glitch_location = 0
 
         # Extract a proxy suggestion prepended to the path
@@ -409,12 +432,16 @@ class _TokenHTTPProxyHandler(http.server.BaseHTTPRequestHandler):
                     self.send_response(status_code, 'Suggested by unit test')
                     self.end_headers()
                     return
+                if command == 'reset-connection':
+                    self._reset_connection()
+                    return
                 # Truncate or pause transmission of the payload after specified bytes
-                glitch = re.match(r'^(truncate|pause)-read-after-(\d+)-bytes$', command)
+                glitch = re.match(r'^(truncate|pause|reset)-read-after-(\d+)-bytes$', command)
                 if glitch:
                     flavour = glitch.group(1)
                     truncate = (flavour == 'truncate')
                     pause = READ_PAUSE if flavour == 'pause' else 0.0
+                    reset = (flavour == 'reset')
                     glitch_location = int(glitch.group(2))
                 else:
                     raise ValueError(f"Unknown command '{command}' "
@@ -474,14 +501,19 @@ class _TokenHTTPProxyHandler(http.server.BaseHTTPRequestHandler):
         # might stop listening if it knows nothing more is coming, like in a PUT response).
         if len(content) == 0:
             return
-        if pause:
-            self.wfile.write(content[:glitch_location])
+        # Write the first half of the data
+        self.wfile.write(content[:glitch_location])
+        if truncate:
+            return
+        elif reset:
+            self._reset_connection()
+            return
+        elif pause:
             # The wfile object should be an unbuffered _SocketWriter but flush anyway
             self.wfile.flush()
             time.sleep(pause)
-            self.wfile.write(content[glitch_location:])
-        else:
-            self.wfile.write(content[:glitch_location] if truncate else content)
+        # Write the rest of the data
+        self.wfile.write(content[glitch_location:])
 
     def log_message(self, format, *args):
         # Get time offset from first of these requests (useful for debugging)
@@ -547,7 +579,8 @@ class TestS3ChunkStoreToken(TestS3ChunkStore):
         elif url != cls.httpd.target:
             raise RuntimeError('Cannot use multiple target URLs with http proxy')
         # The token authorises the standard bucket and anything starting with PREFIX
-        token = encode_jwt({'prefix': [BUCKET, PREFIX]})
+        # (as well as suggestions prepended to the path)
+        token = encode_jwt({'prefix': [BUCKET, PREFIX, 'please']})
         kwargs.setdefault('token', token)
         return super().prepare_store_args(cls.proxy_url, credentials=None, **kwargs)
 
@@ -570,68 +603,160 @@ class TestS3ChunkStoreToken(TestS3ChunkStore):
         self.store.put_chunk(array_name, slices, chunk)
         return chunk, slices, self.store.join(array_name, suggestion)
 
-    @pytest.mark.expected_duration(0.9)
+    # 50x STATUSES
+    #
+    # With the RETRY settings of 2 status retries, backoff factor of 0.1 s
+    # and SUGGESTED_STATUS_DELAY of 0.1 s we get the following timeline
+    # (indexed by seconds):
+    # 0.0 - access chunk for the first time
+    # 0.1 - response is 500, immediately try again (retry #1)
+    # 0.2 - response is 500, back off for 2 * 0.1 seconds
+    # 0.4 - retry #2 (the final attempt) - server should now be fixed
+    # 0.4 - success!
+
+    @pytest.mark.expected_duration(0.1)
+    def test_server_error(self):
+        suggestion = 'please-respond-with-500-for-0.2-seconds'
+        chunk, slices, array_name = self._put_chunk(suggestion)
+        chunk_name, _ = self.store.chunk_metadata(array_name, slices, dtype=chunk.dtype)
+        url = self.store._chunk_url(chunk_name)
+        response = self.store.request('GET', url, ignored_errors=(500,), retries=0, stream=True)
+        assert response.status_code == 500
+
+    @pytest.mark.expected_duration(0.4)
     def test_recover_from_server_errors(self):
-        chunk, slices, array_name = self._put_chunk(
-            'please-respond-with-500-for-0.8-seconds')
-        # With the RETRY settings of 3 status retries, backoff factor of 0.1 s
-        # and SUGGESTED_STATUS_DELAY of 0.1 s we get the following timeline
-        # (indexed by seconds):
-        # 0.0 - access chunk for the first time
-        # 0.1 - response is 500, immediately try again (retry #1)
-        # 0.2 - response is 500, back off for 2 * 0.1 seconds
-        # 0.4 - retry #2
-        # 0.5 - response is 500, back off for 4 * 0.1 seconds
-        # 0.9 - retry #3 (the final attempt) - server should now be fixed
-        # 0.9 - success!
-        self.store.get_chunk(array_name, slices, chunk.dtype)
-
-    @pytest.mark.expected_duration(1.0)
+        suggestion = 'please-respond-with-500-for-0.3-seconds'
+        chunk, slices, array_name = self._put_chunk(suggestion)
+        chunk_retrieved = self.store.get_chunk(array_name, slices, chunk.dtype)
+        assert_array_equal(chunk_retrieved, chunk, 'Bad chunk after server error')
+
+    @pytest.mark.expected_duration(0.5)
     def test_persistent_server_errors(self):
-        chunk, slices, array_name = self._put_chunk(
-            'please-respond-with-502-for-1.2-seconds')
-        # After 0.9 seconds the client gives up and returns with failure 0.1 s later
+        suggestion = 'please-respond-with-502-for-0.7-seconds'
+        chunk, slices, array_name = self._put_chunk(suggestion)
+        # After 0.4 seconds the client gives up and returns with failure 0.1 s later
         with pytest.raises(ChunkNotFound):
             self.store.get_chunk(array_name, slices, chunk.dtype)
 
-    @pytest.mark.expected_duration(0.6)
-    def test_recover_from_read_truncated_within_npy_header(self):
-        chunk, slices, array_name = self._put_chunk(
-            'please-truncate-read-after-60-bytes-for-0.4-seconds')
-        # With the RETRY settings of 3 status retries and backoff factor of 0.1 s
-        # we get the following timeline (indexed by seconds):
-        # 0.0 - access chunk for the first time
-        # 0.0 - response is 200 but truncated, immediately try again (retry #1)
-        # 0.0 - response is 200 but truncated, back off for 2 * 0.1 seconds
-        # 0.2 - retry #2, response is 200 but truncated, back off for 4 * 0.1 seconds
-        # 0.6 - retry #3 (the final attempt) - server should now be fixed
-        # 0.6 - success!
-        chunk_retrieved = self.store.get_chunk(array_name, slices, chunk.dtype)
-        assert_array_equal(chunk_retrieved, chunk, 'Truncated read not recovered')
+    # TRUNCATED READS
+    #
+    # With the RETRY settings of 3 read retries and backoff factor of 0.1 s
+    # we get the following timeline (indexed by seconds):
+    # 0.0 - access chunk for the first time
+    # 0.0 - response is 200 but truncated, immediately try again (retry #1)
+    # 0.0 - response is 200 but truncated, back off for 2 * 0.1 seconds
+    # 0.2 - retry #2, response is 200 but truncated, back off for 4 * 0.1 seconds
+    # 0.6 - retry #3 (the final attempt) - server should now be fixed
+    # 0.6 - success!
+
+    # The NPY file has a 128-byte header, followed by the array data itself.
+    # Check both parts, since they are read somewhat differently (read vs readinto).
+    @pytest.mark.parametrize('nbytes', [60, 129])
+    @pytest.mark.expected_duration(0.0)
+    def test_truncated_read(self, nbytes):
+        suggestion = f'please-truncate-read-after-{nbytes}-bytes-for-0.1-seconds'
+        chunk, slices, array_name = self._put_chunk(suggestion)
+        self.store.retries = Retry(0)
+        with pytest.raises(S3ServerGlitch) as excinfo:
+            self.store.get_chunk(array_name, slices, chunk.dtype)
+        bytes_left = 128 + chunk.nbytes - nbytes
+        excinfo.match(fr'IncompleteRead\({nbytes} bytes read, {bytes_left} more expected\)')
 
+    @pytest.mark.parametrize('nbytes', [60, 129])
     @pytest.mark.expected_duration(0.6)
-    def test_recover_from_read_truncated_within_npy_array(self):
-        chunk, slices, array_name = self._put_chunk(
-            'please-truncate-read-after-129-bytes-for-0.4-seconds')
+    def test_recover_from_truncated_read(self, nbytes):
+        suggestion = f'please-truncate-read-after-{nbytes}-bytes-for-0.4-seconds'
+        chunk, slices, array_name = self._put_chunk(suggestion)
         chunk_retrieved = self.store.get_chunk(array_name, slices, chunk.dtype)
-        assert_array_equal(chunk_retrieved, chunk, 'Truncated read not recovered')
+        assert_array_equal(chunk_retrieved, chunk, 'Bad chunk after truncated read')
 
     @pytest.mark.expected_duration(0.6)
     def test_persistent_truncated_reads(self):
-        chunk, slices, array_name = self._put_chunk(
-            'please-truncate-read-after-60-bytes-for-0.8-seconds')
+        suggestion = 'please-truncate-read-after-60-bytes-for-0.8-seconds'
+        chunk, slices, array_name = self._put_chunk(suggestion)
         # After 0.6 seconds the client gives up
         with pytest.raises(ChunkNotFound):
             self.store.get_chunk(array_name, slices, chunk.dtype)
 
-    @pytest.mark.expected_duration(READ_PAUSE)
-    def test_handle_read_paused_within_npy_header(self):
-        chunk, slices, array_name = self._put_chunk('please-pause-read-after-60-bytes')
+    def test_rdb_support_recover_from_truncated_reads(self):
+        super().test_rdb_support('please-truncate-read-after-1000-bytes-for-0.4-seconds')
+
+    def test_rdb_support_persistent_truncated_reads(self):
+        with pytest.raises(DataSourceNotFound) as excinfo:
+            super().test_rdb_support('please-truncate-read-after-1000-bytes-for-0.8-seconds')
+        excinfo.match(r'IncompleteRead\(1000 bytes read')
+
+    @pytest.mark.expected_duration(0.6)
+    def test_recover_from_reset_connections(self):
+        suggestion = 'please-reset-read-after-129-bytes-for-0.4-seconds'
+        chunk, slices, array_name = self._put_chunk(suggestion)
         chunk_retrieved = self.store.get_chunk(array_name, slices, chunk.dtype)
-        assert_array_equal(chunk_retrieved, chunk, 'Paused read failed')
+        assert_array_equal(chunk_retrieved, chunk, 'Bad chunk after reset connection')
+
+    @pytest.mark.expected_duration(0.6)
+    def test_persistent_reset_connections(self):
+        suggestion = 'please-reset-read-after-129-bytes-for-0.8-seconds'
+        chunk, slices, array_name = self._put_chunk(suggestion)
+        with pytest.raises(ChunkNotFound) as excinfo:
+            self.store.get_chunk(array_name, slices, chunk.dtype)
+        assert isinstance(excinfo.value, S3ServerGlitch)
 
+    @pytest.mark.expected_duration(0.6)
+    def test_persistent_early_reset_connections(self):
+        suggestion = 'please-reset-connection-for-0.8-seconds'
+        chunk, slices, array_name = self._put_chunk(suggestion)
+        with pytest.raises(StoreUnavailable):
+            self.store.get_chunk(array_name, slices, chunk.dtype)
+
+    @pytest.mark.parametrize('nbytes', [60, 129])  # check both NPY header and array itself
     @pytest.mark.expected_duration(READ_PAUSE)
-    def test_handle_read_paused_within_npy_array(self):
-        chunk, slices, array_name = self._put_chunk('please-pause-read-after-129-bytes')
+    def test_handle_paused_read(self, nbytes):
+        suggestion = f'please-pause-read-after-{nbytes}-bytes'
+        chunk, slices, array_name = self._put_chunk(suggestion)
+        chunk_retrieved = self.store.get_chunk(array_name, slices, chunk.dtype)
+        assert_array_equal(chunk_retrieved, chunk, 'Bad chunk after paused read')
+
+    # SOCKET TIMEOUTS
+    #
+    # With a read timeout of 0.09 seconds and the RETRY settings of 3 read retries and
+    # backoff factor of 0.1 s, we get the following timeline (indexed by seconds):
+    # 0.00 - access chunk for the first time
+    # 0.09 - response is 200 but socket times out after 0.09 seconds, immediately try again
+    # 0.18 - retry #1, response is 200 but socket stalls, back off for 2 * 0.1 seconds
+    # 0.47 - retry #2, response is 200 but socket stalls, back off for 4 * 0.1 seconds
+    # 0.87 - retry #3 (the final attempt) - server should now be fixed
+    # 0.87 - success!
+
+    @pytest.mark.parametrize('nbytes', [60, 129])  # check both NPY header and array itself
+    @pytest.mark.expected_duration(0.87)
+    def test_recover_from_socket_timeout(self, nbytes):
+        suggestion = f'please-pause-read-after-{nbytes}-bytes-for-0.8-seconds'
+        chunk, slices, array_name = self._put_chunk(suggestion)
+        self.store.timeout = (5.0, 0.09)
         chunk_retrieved = self.store.get_chunk(array_name, slices, chunk.dtype)
-        assert_array_equal(chunk_retrieved, chunk, 'Paused read failed')
+        assert_array_equal(chunk_retrieved, chunk, 'Bad chunk after socket timeout')
+
+    @pytest.mark.expected_duration(0.96)
+    def test_persistent_socket_timeouts(self):
+        suggestion = 'please-pause-read-after-129-bytes-for-1.0-seconds'
+        chunk, slices, array_name = self._put_chunk(suggestion)
+        self.store.timeout = (5.0, 0.09)
+        # The final retry starts at 0.87 seconds and the client gives up 0.09 seconds later
+        with pytest.raises(ChunkNotFound):
+            self.store.get_chunk(array_name, slices, chunk.dtype)
+
+    # XXX Check whether the 'pause' condition (socket timeouts) can be tested
+    # with _read_object, which preloads the content due to stream=False.
+    @pytest.mark.parametrize('condition', ['truncate', 'reset'])
+    @pytest.mark.expected_duration(0.6)
+    def test_non_streaming_request_recovery_from_glitches(self, condition):
+        data = b'x' * 1000
+        cbid = PREFIX
+        path = self.store.join(cbid, f'test_recovery_from_{condition}.bin')
+        url = urllib.parse.urljoin(self.store_url, path)
+        self.store.create_array(cbid)
+        self.store.request('PUT', url, data=data)
+        suggestion = f'please-{condition}-read-after-400-bytes-for-0.52-seconds'
+        url = urllib.parse.urljoin(self.store_url, self.store.join(suggestion, path))
+        retrieved_data = self.store.request('GET', url, process=_read_object)
+        assert retrieved_data == data