From b53e63a03c28585b7cb858d74388b9a4fc4fae95 Mon Sep 17 00:00:00 2001 From: Kevin Smith Date: Mon, 9 Nov 2026 08:48:52 -0500 Subject: [PATCH 1/7] Keep live v2 test clusters out of AWS us-east-1 Since 2026-11-04 every S-00 cluster the nightly coverage run created in AWS us-east-1 stayed PENDING past the 1200s wait, failing TestClusterFusionCreateDrop::test_create_drop_cluster in five of the last six runs. That test always took us_regions[0], which the v2 API lists as us-east-1, so it never escaped the bad region; clusters elsewhere in the US came up normally. Move the US region filter into utils.us_regions(), drop AVOIDED_REGIONS from it, and use it for both the Fusion cluster fixture and TestCluster. test_create_drop_cluster now picks a random region like the fixtures do. Co-Authored-By: Claude Opus 5.5 --- singlestoredb/tests/test_fusion.py | 9 ++--- singlestoredb/tests/test_management_utils.py | 36 ++++++++++++++++++++ singlestoredb/tests/test_management_v2.py | 10 +----- singlestoredb/tests/utils.py | 26 ++++++++++++++ 4 files changed, 65 insertions(+), 16 deletions(-) diff --git a/singlestoredb/tests/test_fusion.py b/singlestoredb/tests/test_fusion.py index fa206ae44..eb9d2cf15 100644 --- a/singlestoredb/tests/test_fusion.py +++ b/singlestoredb/tests/test_fusion.py @@ -1501,12 +1501,7 @@ def setUpClass(cls): mgr = s2.manage_clusters(version='v2') cls.manager = mgr - cls.us_regions = [ - x for x in mgr.regions - if 'US' in x.name or 'us-' in (x.region_name or '') - ] - if not cls.us_regions: - raise unittest.SkipTest('No US regions reported by the v2 API') + cls.us_regions = utils.us_regions(mgr) cls.project_id = cls._project_id(mgr) @@ -1765,7 +1760,7 @@ def test_create_drop_cluster(self): mgr = type(self).manager name = f'd-fusion-cluster-{self.id}' - region = type(self).us_regions[0] + region = random.choice(type(self).us_regions) try: self.cur.execute( diff --git a/singlestoredb/tests/test_management_utils.py b/singlestoredb/tests/test_management_utils.py index 1345390d8..33198a542 100644 --- a/singlestoredb/tests/test_management_utils.py +++ b/singlestoredb/tests/test_management_utils.py @@ -26,6 +26,7 @@ from singlestoredb.tests.utils import admin_password from singlestoredb.tests.utils import counting_file_space from singlestoredb.tests.utils import counting_stage +from singlestoredb.tests.utils import us_regions TEST_DIR = pathlib.Path(os.path.dirname(__file__)) @@ -716,6 +717,41 @@ def test_listdir_style_suffix(self): assert normalize_remote_path('/', strip_leading=True) + '/' == '/' +class TestUSRegions(unittest.TestCase): + """Coverage for the region filter the live v2 suites deploy through.""" + + @staticmethod + def _region(name, provider, region_name): + return SimpleNamespace( + name=name, provider=provider, region_name=region_name, + ) + + def test_keeps_us_regions_and_drops_avoided_ones(self): + mgr = SimpleNamespace( + regions=[ + self._region('US East 1 (N. Virginia)', 'AWS', 'us-east-1'), + self._region('US East 2 (Ohio)', 'AWS', 'us-east-2'), + self._region('US East 1 (S. Carolina)', 'GCP', 'us-east1'), + self._region('East US 2 (Virginia)', 'Azure', 'eastus2'), + self._region('Europe West 1 (Ireland)', 'AWS', 'eu-west-1'), + ], + ) + got = [(x.provider, x.region_name) for x in us_regions(mgr)] + self.assertEqual( + got, + [('AWS', 'us-east-2'), ('GCP', 'us-east1'), ('Azure', 'eastus2')], + ) + + def test_skips_when_only_avoided_regions_are_left(self): + mgr = SimpleNamespace( + regions=[ + self._region('US East 1 (N. Virginia)', 'AWS', 'us-east-1'), + ], + ) + with self.assertRaises(unittest.SkipTest): + us_regions(mgr) + + class TestSecretFromDictTimestamps(unittest.TestCase): """ Coverage for ``Secret.from_dict`` running its timestamp fields diff --git a/singlestoredb/tests/test_management_v2.py b/singlestoredb/tests/test_management_v2.py index 6618ccd60..b84d7050d 100644 --- a/singlestoredb/tests/test_management_v2.py +++ b/singlestoredb/tests/test_management_v2.py @@ -74,14 +74,6 @@ def shared_database_name(s): return re.sub(r'[^\w]', '', s).replace('-', '_').lower() -def _us_regions(manager): - """Return the US regions a v2 manager reports, or skip the test.""" - out = [x for x in manager.regions if 'US' in x.name or 'us-' in x.name] - if not out: - raise unittest.SkipTest('No US regions reported by the v2 API') - return out - - def _project_id(manager): """ Return the project ID the live v2 suites deploy into, or skip the test. @@ -1320,7 +1312,7 @@ class TestCluster(unittest.TestCase): def setUpClass(cls): cls.manager = s2.manage_clusters(version='v2') - us_regions = _us_regions(cls.manager) + us_regions = utils.us_regions(cls.manager) name = clean_name(secrets.token_urlsafe(20)[:20]) region = random.choice(us_regions) diff --git a/singlestoredb/tests/utils.py b/singlestoredb/tests/utils.py index f0b22e91f..b94c46bc9 100644 --- a/singlestoredb/tests/utils.py +++ b/singlestoredb/tests/utils.py @@ -376,6 +376,32 @@ def admin_password(length: int = 24) -> str: #: and the starter deployments accept no such argument. DEPLOYMENT_EXPIRES_AT = '2h' +#: (provider, region_name) pairs the live v2 suites never deploy into. +#: +#: AWS us-east-1: from 2026-11-04 every S-00 cluster the nightly coverage run +#: created there sat in PENDING past the 1200s wait, while clusters in the +#: other US regions came up in minutes. Nothing the suites assert depends on +#: the region, so leaving it out costs no coverage. +AVOIDED_REGIONS = {('AWS', 'us-east-1')} + + +def us_regions(manager: Any) -> List[Any]: + """ + Return the US regions a v2 manager reports, less ``AVOIDED_REGIONS``. + + Raises ``unittest.SkipTest`` if none are left. + + """ + out = [ + x for x in manager.regions + if ('US' in x.name or 'us-' in (x.region_name or '')) + and (x.provider, x.region_name) not in AVOIDED_REGIONS + ] + if not out: + raise unittest.SkipTest('No usable US regions reported by the v2 API') + return out + + #: (owner, label, object) for every deployment created so far and not yet #: swept. The owner is the test class that was running at creation time, so #: a class's leftovers can be dropped when the run leaves that class rather From a30c070f51683ad34e6cde99627c217a10a6f232 Mon Sep 17 00:00:00 2001 From: Kevin Smith Date: Mon, 9 Nov 2026 09:22:37 -0500 Subject: [PATCH 2/7] Route the shared cluster pool through us_regions() shared_clusters() kept its own copy of the US region filter, so the pool the Stage, Job and Fusion suites borrow from could still deploy into AWS us-east-1 and hit the same PENDING timeout. Use utils.us_regions(), keeping the cached _pool_skip, and cover the avoided-region path in the pool tests. Co-Authored-By: Claude Opus 5.5 --- singlestoredb/tests/test_management_utils.py | 16 ++++++++++++++++ singlestoredb/tests/utils.py | 14 ++++++-------- 2 files changed, 22 insertions(+), 8 deletions(-) diff --git a/singlestoredb/tests/test_management_utils.py b/singlestoredb/tests/test_management_utils.py index 33198a542..f472ba599 100644 --- a/singlestoredb/tests/test_management_utils.py +++ b/singlestoredb/tests/test_management_utils.py @@ -2238,6 +2238,7 @@ class Region: def __init__(self, name): self.name = name self.region_name = name + self.provider = 'AWS' class Project: def __init__(self, edition): @@ -2356,6 +2357,21 @@ def test_no_us_region_skips_rather_than_failing(self): self.assertEqual(self.created, []) + def test_pool_never_deploys_to_an_avoided_region(self): + with self._patched(regions=('us-east-1', 'us-east-2')): + self.utils.shared_clusters(3) + + self.assertEqual( + [kwargs['region'].region_name for _, _, kwargs in self.created], + ['us-east-2'] * 3, + ) + + def test_only_avoided_regions_skips_rather_than_failing(self): + with self._patched(regions=('us-east-1',)): + with self.assertRaises(unittest.SkipTest): + self.utils.shared_clusters(1) + self.assertEqual(self.created, []) + def test_no_standard_project_skips_rather_than_failing(self): with self._patched(projects=('SHARED',)): with self.assertRaises(unittest.SkipTest) as cm: diff --git a/singlestoredb/tests/utils.py b/singlestoredb/tests/utils.py index b94c46bc9..0e51a8811 100644 --- a/singlestoredb/tests/utils.py +++ b/singlestoredb/tests/utils.py @@ -1140,13 +1140,11 @@ def setUpClass(cls): # not follow the management.version option out of v2 either. mgr = s2.manage_clusters(version='v2') - us_regions = [ - x for x in mgr.regions - if 'US' in x.name or 'us-' in (x.region_name or '') - ] - if not us_regions: - _pool_skip = 'No US regions reported by the v2 API' - raise unittest.SkipTest(_pool_skip) + try: + regions = us_regions(mgr) + except unittest.SkipTest as exc: + _pool_skip = str(exc) + raise project_id = os.environ.get('SINGLESTOREDB_TEST_PROJECT') if not project_id: @@ -1172,7 +1170,7 @@ def setUpClass(cls): _pool.append( mgr.create_cluster( f'cl-test-shared-{len(_pool)}-{_pool_id}', - region=random.choice(us_regions), + region=random.choice(regions), size='S-00', # The v2 suites that deploy their own ask for this, and a # pool cluster stands in for those, so it has to be at From 1c24545495e63fb44633548e047efa48edc878aa Mon Sep 17 00:00:00 2001 From: Kevin Smith Date: Mon, 9 Nov 2026 12:35:08 -0500 Subject: [PATCH 3/7] Stop replaying PUTs; check whether a timed-out upload landed Every PUT the management API takes is a Files or Stage create. When the Files API outlasted the 180s read timeout, urllib3 replayed the upload, and the replay of one that had landed failed with 409 "File with the same name already exists" -- the TestFilesFusion failures in coverage. Drop PUT from RETRY_METHODS. FileLocation._put_file now handles a transport failure on an upload by asking whether the file is there: _upload has just seen the path absent, so a file at it is the one this upload wrote, and the failure is dropped. Otherwise the original error is raised. Co-Authored-By: Claude Opus 5.5 --- singlestoredb/management/files.py | 50 ++++++++++++++++++++++++++--- singlestoredb/management/manager.py | 7 +++- singlestoredb/management/stage.py | 6 +--- 3 files changed, 52 insertions(+), 11 deletions(-) diff --git a/singlestoredb/management/files.py b/singlestoredb/management/files.py index eacf6c7bb..710e38306 100644 --- a/singlestoredb/management/files.py +++ b/singlestoredb/management/files.py @@ -17,6 +17,8 @@ from typing import overload from typing import Union +import requests + from .. import config from ..exceptions import ManagementError from .manager import Manager @@ -498,6 +500,48 @@ def _info_or_none(self, path: PathLike) -> Optional[FilesObject]: return None raise + def _put_file( + self, + route: str, + path: PathLike, + content: Union[str, bytes, io.IOBase], + ) -> None: + """ + Write ``content`` to ``route``, surviving a lost response. + + A transport failure -- most often the read timeout on a slow Files + API -- does not say whether the upload landed, and the route only + creates, so it cannot be replayed. The caller has just seen ``path`` + absent, so a file at it now is the one this upload wrote, and the + failure is dropped. Anything else, including a failure of that check, + raises the original error. + + Parameters + ---------- + route : str + Request path of the upload + path : Path or str + Path of the file, for the existence check + content : str or bytes or file-like + Content to upload + + """ + try: + self._manager._put( + route, + files={'file': content}, + headers={'Content-Type': None}, + ) + except ManagementError as exc: + if not isinstance(exc.__cause__, requests.exceptions.RequestException): + raise + try: + landed = self._info_or_none(path) is not None + except ManagementError: + landed = False + if not landed: + raise exc + @abstractmethod def exists(self, path: PathLike) -> bool: pass @@ -880,11 +924,7 @@ def _upload( raise IsADirectoryError('file path is a directory') self._manager._delete(f'files/fs/{self._location}/{path}') - self._manager._put( - f'files/fs/{self._location}/{path}', - files={'file': content}, - headers={'Content-Type': None}, - ) + self._put_file(f'files/fs/{self._location}/{path}', path, content) return self.info(path) if fetch_info else None diff --git a/singlestoredb/management/manager.py b/singlestoredb/management/manager.py index c20f571de..9d0e5f327 100644 --- a/singlestoredb/management/manager.py +++ b/singlestoredb/management/manager.py @@ -52,7 +52,12 @@ def set_organization(kwargs: Dict[str, Any]) -> None: #: #: :func:`retry_on_lock` is the one POST replay, keyed on an error message this #: policy never sees. -RETRY_METHODS = frozenset(['GET', 'HEAD', 'OPTIONS', 'PUT', 'DELETE']) +#: +#: PUT is absent too. Every PUT the API takes is a Files or Stage create, which +#: answers a replay of one that landed with 409 "File with the same name +#: already exists". :meth:`FileLocation._put_file` asks whether the file is +#: there instead. +RETRY_METHODS = frozenset(['GET', 'HEAD', 'OPTIONS', 'DELETE']) #: Status codes worth retrying. These are the transient ones; a 4xx other #: than 429 is a client error that will fail again identically. diff --git a/singlestoredb/management/stage.py b/singlestoredb/management/stage.py index ac724b35e..12cd8f836 100644 --- a/singlestoredb/management/stage.py +++ b/singlestoredb/management/stage.py @@ -298,11 +298,7 @@ def _upload( ) self._manager._delete(self._fs_path(stage_path)) - self._manager._put( - self._fs_path(stage_path), - files={'file': content}, - headers={'Content-Type': None}, - ) + self._put_file(self._fs_path(stage_path), stage_path, content) return self.info(stage_path) if fetch_info else None From b8f78e885230fb8e9646e11f0fd04ce5838418dc Mon Sep 17 00:00:00 2001 From: Kevin Smith Date: Mon, 9 Nov 2026 13:30:47 -0500 Subject: [PATCH 4/7] Check whether a timed-out Stage mkdir landed [skip ci] Stage.mkdir is a PUT too, so dropping PUT from RETRY_METHODS left it with no recovery from a lost response. Generalize _put_file into FileLocation._create and route mkdir through it: mkdir has just seen the path absent or removed it, so a folder there now is the one it made. Fix CountingManager._put's folder normalization along the way. re.sub(r'/*$', '/', 'd/') also matches the empty string at the end and gave 'd//', so a folder the fake had created could not be looked up. Co-Authored-By: Claude Opus 5.5 --- singlestoredb/management/files.py | 33 ++++++++++++----------------- singlestoredb/management/manager.py | 4 ++-- singlestoredb/management/stage.py | 10 +++++---- singlestoredb/tests/utils.py | 2 +- 4 files changed, 23 insertions(+), 26 deletions(-) diff --git a/singlestoredb/management/files.py b/singlestoredb/management/files.py index 710e38306..2b75044dd 100644 --- a/singlestoredb/management/files.py +++ b/singlestoredb/management/files.py @@ -500,38 +500,29 @@ def _info_or_none(self, path: PathLike) -> Optional[FilesObject]: return None raise - def _put_file( - self, - route: str, - path: PathLike, - content: Union[str, bytes, io.IOBase], - ) -> None: + def _create(self, route: str, path: PathLike, **kwargs: Any) -> None: """ - Write ``content`` to ``route``, surviving a lost response. + PUT to ``route`` to create ``path``, surviving a lost response. A transport failure -- most often the read timeout on a slow Files - API -- does not say whether the upload landed, and the route only + API -- does not say whether the create landed, and the route only creates, so it cannot be replayed. The caller has just seen ``path`` - absent, so a file at it now is the one this upload wrote, and the + absent, so an object at it now is the one this request made, and the failure is dropped. Anything else, including a failure of that check, raises the original error. Parameters ---------- route : str - Request path of the upload + Request path of the create path : Path or str - Path of the file, for the existence check - content : str or bytes or file-like - Content to upload + Path of the file or folder, for the existence check + **kwargs : keyword arguments, optional + Keyword arguments to add to the PUT request """ try: - self._manager._put( - route, - files={'file': content}, - headers={'Content-Type': None}, - ) + self._manager._put(route, **kwargs) except ManagementError as exc: if not isinstance(exc.__cause__, requests.exceptions.RequestException): raise @@ -924,7 +915,11 @@ def _upload( raise IsADirectoryError('file path is a directory') self._manager._delete(f'files/fs/{self._location}/{path}') - self._put_file(f'files/fs/{self._location}/{path}', path, content) + self._create( + f'files/fs/{self._location}/{path}', path, + files={'file': content}, + headers={'Content-Type': None}, + ) return self.info(path) if fetch_info else None diff --git a/singlestoredb/management/manager.py b/singlestoredb/management/manager.py index 9d0e5f327..e1f0c5897 100644 --- a/singlestoredb/management/manager.py +++ b/singlestoredb/management/manager.py @@ -55,8 +55,8 @@ def set_organization(kwargs: Dict[str, Any]) -> None: #: #: PUT is absent too. Every PUT the API takes is a Files or Stage create, which #: answers a replay of one that landed with 409 "File with the same name -#: already exists". :meth:`FileLocation._put_file` asks whether the file is -#: there instead. +#: already exists". :meth:`FileLocation._create` asks whether the file or +#: folder is there instead. RETRY_METHODS = frozenset(['GET', 'HEAD', 'OPTIONS', 'DELETE']) #: Status codes worth retrying. These are the transient ones; a 4xx other diff --git a/singlestoredb/management/stage.py b/singlestoredb/management/stage.py index 12cd8f836..b03fe8a7e 100644 --- a/singlestoredb/management/stage.py +++ b/singlestoredb/management/stage.py @@ -298,7 +298,11 @@ def _upload( ) self._manager._delete(self._fs_path(stage_path)) - self._put_file(self._fs_path(stage_path), stage_path, content) + self._create( + self._fs_path(stage_path), stage_path, + files={'file': content}, + headers={'Content-Type': None}, + ) return self.info(stage_path) if fetch_info else None @@ -326,9 +330,7 @@ def mkdir(self, stage_path: PathLike, overwrite: bool = False) -> FilesObject: self.remove(stage_path) - self._manager._put( - self._fs_path(stage_path) + '?isFile=false', - ) + self._create(self._fs_path(stage_path) + '?isFile=false', stage_path) return self.info(stage_path) diff --git a/singlestoredb/tests/utils.py b/singlestoredb/tests/utils.py index 0e51a8811..b6d17e605 100644 --- a/singlestoredb/tests/utils.py +++ b/singlestoredb/tests/utils.py @@ -1282,7 +1282,7 @@ def _get(self, path: Any, params: Any = None, **kwargs: Any) -> Any: def _put(self, path: Any, **kwargs: Any) -> Any: key = self._key(path) if 'isFile=false' in str(path): - key = re.sub(r'/*$', r'/', key) + key = key.rstrip('/') + '/' self.calls.append(('PUT', key)) self.existing.add(key) return SimpleNamespace( From b40f12f67252b2a8c86a62aaca61a30808eba959 Mon Sep 17 00:00:00 2001 From: Kevin Smith Date: Mon, 9 Nov 2026 13:54:53 -0500 Subject: [PATCH 5/7] Only take a lost create as landed when the request was sent [skip ci] Review on #160: an object found at the path after any transport failure was taken as this client's, including when the PUT never reached the server and another writer created the path meanwhile. _create now probes only when the response was lost -- a read timeout, a connection dropped while waiting, or a gateway 502/504 -- and re-raises a connect timeout, a failed connect or a 503 as before. An upload must also find the size it sent. The remaining window, another client writing the same path at the same size inside the timeout, is documented; comparing content would mean downloading from the API that has just failed to answer. Co-Authored-By: Claude Opus 5.5 --- singlestoredb/management/files.py | 86 +++++++++++++++++++++++++++---- singlestoredb/management/stage.py | 2 + 2 files changed, 77 insertions(+), 11 deletions(-) diff --git a/singlestoredb/management/files.py b/singlestoredb/management/files.py index 2b75044dd..462bc4460 100644 --- a/singlestoredb/management/files.py +++ b/singlestoredb/management/files.py @@ -18,6 +18,7 @@ from typing import Union import requests +import urllib3 from .. import config from ..exceptions import ManagementError @@ -358,6 +359,54 @@ class FilesObjectBytesReader(io.BytesIO): """BytesIO wrapper for reading from FileLocation.""" +def _response_was_lost(exc: ManagementError) -> bool: + """ + Did the request reach the server and only its response go missing? + + A read timeout, a connection dropped while waiting for the response, or a + gateway 502/504 for a request it forwarded: the server may have acted. A + connect timeout or a connection that never opened sent nothing, and a 503 + is a refusal, so neither can have created anything. + + """ + if exc.errno in (502, 504): + return True + cause = exc.__cause__ + if isinstance( + cause, + (requests.exceptions.ReadTimeout, requests.exceptions.ChunkedEncodingError), + ): + return True + # requests wraps a dropped response's ProtocolError directly; a failed + # connect arrives wrapped in MaxRetryError instead + if isinstance(cause, requests.exceptions.ConnectionError) and cause.args: + return isinstance(cause.args[0], urllib3.exceptions.ProtocolError) + return False + + +def _content_size(content: Union[str, bytes, io.IOBase]) -> Optional[int]: + """ + Return the number of bytes an upload of ``content`` sends, if known. + + Called before the upload, which consumes a stream. A stream that cannot + seek has no size to offer, and a text stream's positions are not bytes. + + """ + if isinstance(content, bytes): + return len(content) + if isinstance(content, str): + # The multipart body carries a str as UTF-8 + return len(content.encode('utf-8')) + if isinstance(content, io.TextIOBase): + return None + if isinstance(content, io.IOBase) and content.seekable(): + pos = content.tell() + end = content.seek(0, io.SEEK_END) + content.seek(pos) + return end - pos + return None + + class FileLocation(ABC): _manager: Manager @@ -500,16 +549,27 @@ def _info_or_none(self, path: PathLike) -> Optional[FilesObject]: return None raise - def _create(self, route: str, path: PathLike, **kwargs: Any) -> None: + def _create( + self, + route: str, + path: PathLike, + size: Optional[int] = None, + **kwargs: Any, + ) -> None: """ PUT to ``route`` to create ``path``, surviving a lost response. - A transport failure -- most often the read timeout on a slow Files - API -- does not say whether the create landed, and the route only - creates, so it cannot be replayed. The caller has just seen ``path`` - absent, so an object at it now is the one this request made, and the - failure is dropped. Anything else, including a failure of that check, - raises the original error. + A lost response -- most often the read timeout on a slow Files API -- + does not say whether the create landed, and the route only creates, so + it cannot be replayed. The caller has just seen ``path`` absent, so an + object at it now, of the ``size`` that was sent, is taken to be the one + this request made, and the failure is dropped. A request that never + reached the server, or a failure of that check, raises the original + error, as does everything else. + + What this cannot rule out is another client creating the same path, + with the same size, inside the timeout. Comparing content would mean + downloading the file from the API that has just failed to answer. Parameters ---------- @@ -517,6 +577,9 @@ def _create(self, route: str, path: PathLike, **kwargs: Any) -> None: Request path of the create path : Path or str Path of the file or folder, for the existence check + size : int, optional + Size in bytes of the upload, checked against the object found. + ``None`` checks existence only. **kwargs : keyword arguments, optional Keyword arguments to add to the PUT request @@ -524,13 +587,13 @@ def _create(self, route: str, path: PathLike, **kwargs: Any) -> None: try: self._manager._put(route, **kwargs) except ManagementError as exc: - if not isinstance(exc.__cause__, requests.exceptions.RequestException): + if not _response_was_lost(exc): raise try: - landed = self._info_or_none(path) is not None + obj = self._info_or_none(path) except ManagementError: - landed = False - if not landed: + obj = None + if obj is None or (size is not None and obj.size != size): raise exc @abstractmethod @@ -917,6 +980,7 @@ def _upload( self._create( f'files/fs/{self._location}/{path}', path, + size=_content_size(content), files={'file': content}, headers={'Content-Type': None}, ) diff --git a/singlestoredb/management/stage.py b/singlestoredb/management/stage.py index b03fe8a7e..986d0b73e 100644 --- a/singlestoredb/management/stage.py +++ b/singlestoredb/management/stage.py @@ -24,6 +24,7 @@ from ..exceptions import ManagementError from ._version_import import _versioned_attr +from .files import _content_size from .files import FileLocation from .files import FilesObject from .files import FilesObjectBytesReader @@ -300,6 +301,7 @@ def _upload( self._create( self._fs_path(stage_path), stage_path, + size=_content_size(content), files={'file': content}, headers={'Content-Type': None}, ) From 2ee89b017fbefdd97743dad8626b618a7700d4ab Mon Sep 17 00:00:00 2001 From: Kevin Smith Date: Tue, 10 Nov 2026 09:23:31 -0500 Subject: [PATCH 6/7] Drop the existence check after a lost upload response [skip ci] Proving that an object found after a lost response is this client's needs either a content comparison, which is a full download for the large Stage files customers upload, or a size check that can still claim another writer's file. Neither is worth carrying client-side; the slow Files API is being taken up server-side instead. PUT stays out of RETRY_METHODS, so a timed-out upload now raises its ReadTimeout rather than the misleading 409 from a replay. Co-Authored-By: Claude Opus 5.5 --- singlestoredb/management/files.py | 103 +--------------------------- singlestoredb/management/manager.py | 4 +- singlestoredb/management/stage.py | 10 +-- 3 files changed, 9 insertions(+), 108 deletions(-) diff --git a/singlestoredb/management/files.py b/singlestoredb/management/files.py index 462bc4460..eacf6c7bb 100644 --- a/singlestoredb/management/files.py +++ b/singlestoredb/management/files.py @@ -17,9 +17,6 @@ from typing import overload from typing import Union -import requests -import urllib3 - from .. import config from ..exceptions import ManagementError from .manager import Manager @@ -359,54 +356,6 @@ class FilesObjectBytesReader(io.BytesIO): """BytesIO wrapper for reading from FileLocation.""" -def _response_was_lost(exc: ManagementError) -> bool: - """ - Did the request reach the server and only its response go missing? - - A read timeout, a connection dropped while waiting for the response, or a - gateway 502/504 for a request it forwarded: the server may have acted. A - connect timeout or a connection that never opened sent nothing, and a 503 - is a refusal, so neither can have created anything. - - """ - if exc.errno in (502, 504): - return True - cause = exc.__cause__ - if isinstance( - cause, - (requests.exceptions.ReadTimeout, requests.exceptions.ChunkedEncodingError), - ): - return True - # requests wraps a dropped response's ProtocolError directly; a failed - # connect arrives wrapped in MaxRetryError instead - if isinstance(cause, requests.exceptions.ConnectionError) and cause.args: - return isinstance(cause.args[0], urllib3.exceptions.ProtocolError) - return False - - -def _content_size(content: Union[str, bytes, io.IOBase]) -> Optional[int]: - """ - Return the number of bytes an upload of ``content`` sends, if known. - - Called before the upload, which consumes a stream. A stream that cannot - seek has no size to offer, and a text stream's positions are not bytes. - - """ - if isinstance(content, bytes): - return len(content) - if isinstance(content, str): - # The multipart body carries a str as UTF-8 - return len(content.encode('utf-8')) - if isinstance(content, io.TextIOBase): - return None - if isinstance(content, io.IOBase) and content.seekable(): - pos = content.tell() - end = content.seek(0, io.SEEK_END) - content.seek(pos) - return end - pos - return None - - class FileLocation(ABC): _manager: Manager @@ -549,53 +498,6 @@ def _info_or_none(self, path: PathLike) -> Optional[FilesObject]: return None raise - def _create( - self, - route: str, - path: PathLike, - size: Optional[int] = None, - **kwargs: Any, - ) -> None: - """ - PUT to ``route`` to create ``path``, surviving a lost response. - - A lost response -- most often the read timeout on a slow Files API -- - does not say whether the create landed, and the route only creates, so - it cannot be replayed. The caller has just seen ``path`` absent, so an - object at it now, of the ``size`` that was sent, is taken to be the one - this request made, and the failure is dropped. A request that never - reached the server, or a failure of that check, raises the original - error, as does everything else. - - What this cannot rule out is another client creating the same path, - with the same size, inside the timeout. Comparing content would mean - downloading the file from the API that has just failed to answer. - - Parameters - ---------- - route : str - Request path of the create - path : Path or str - Path of the file or folder, for the existence check - size : int, optional - Size in bytes of the upload, checked against the object found. - ``None`` checks existence only. - **kwargs : keyword arguments, optional - Keyword arguments to add to the PUT request - - """ - try: - self._manager._put(route, **kwargs) - except ManagementError as exc: - if not _response_was_lost(exc): - raise - try: - obj = self._info_or_none(path) - except ManagementError: - obj = None - if obj is None or (size is not None and obj.size != size): - raise exc - @abstractmethod def exists(self, path: PathLike) -> bool: pass @@ -978,9 +880,8 @@ def _upload( raise IsADirectoryError('file path is a directory') self._manager._delete(f'files/fs/{self._location}/{path}') - self._create( - f'files/fs/{self._location}/{path}', path, - size=_content_size(content), + self._manager._put( + f'files/fs/{self._location}/{path}', files={'file': content}, headers={'Content-Type': None}, ) diff --git a/singlestoredb/management/manager.py b/singlestoredb/management/manager.py index e1f0c5897..230d71211 100644 --- a/singlestoredb/management/manager.py +++ b/singlestoredb/management/manager.py @@ -55,8 +55,8 @@ def set_organization(kwargs: Dict[str, Any]) -> None: #: #: PUT is absent too. Every PUT the API takes is a Files or Stage create, which #: answers a replay of one that landed with 409 "File with the same name -#: already exists". :meth:`FileLocation._create` asks whether the file or -#: folder is there instead. +#: already exists" -- hiding the timeout that caused the replay. A lost +#: response on an upload is raised as it is, and the caller decides. RETRY_METHODS = frozenset(['GET', 'HEAD', 'OPTIONS', 'DELETE']) #: Status codes worth retrying. These are the transient ones; a 4xx other diff --git a/singlestoredb/management/stage.py b/singlestoredb/management/stage.py index 986d0b73e..ac724b35e 100644 --- a/singlestoredb/management/stage.py +++ b/singlestoredb/management/stage.py @@ -24,7 +24,6 @@ from ..exceptions import ManagementError from ._version_import import _versioned_attr -from .files import _content_size from .files import FileLocation from .files import FilesObject from .files import FilesObjectBytesReader @@ -299,9 +298,8 @@ def _upload( ) self._manager._delete(self._fs_path(stage_path)) - self._create( - self._fs_path(stage_path), stage_path, - size=_content_size(content), + self._manager._put( + self._fs_path(stage_path), files={'file': content}, headers={'Content-Type': None}, ) @@ -332,7 +330,9 @@ def mkdir(self, stage_path: PathLike, overwrite: bool = False) -> FilesObject: self.remove(stage_path) - self._create(self._fs_path(stage_path) + '?isFile=false', stage_path) + self._manager._put( + self._fs_path(stage_path) + '?isFile=false', + ) return self.info(stage_path) From 0f5b230faf22ac2cfab30eecfd0d650d65f3633f Mon Sep 17 00:00:00 2001 From: Kevin Smith Date: Tue, 10 Nov 2026 15:27:02 -0500 Subject: [PATCH 7/7] Trigger PR checks Co-Authored-By: Claude Opus 5.5