COPY TO 's3://...' hangs forever when a CreateMultipartUpload fails: uploads_in_progress is leaked in FlushBuffer

## Summary

`S3MultiPartUpload::FlushBuffer` increments `uploads_in_progress`, then — before creating the uploader thread that is responsible for decrementing it again — performs a network call that can throw. When it throws, nothing is left that will ever decrement the counter, and `S3FileHandle` destruction blocks in an untimed `condition_variable::wait` for it to reach zero.

Three facts from the reproduction below, which is a single self-contained script:

- **No `PUT ?partNumber=` is ever sent.** The uploader thread — the only thing that decrements the counter — was never created.
- **The failing request is not retried.** In the log below the request right after the 502 is the cleanup `HEAD`, not another `POST ?uploads=` — so raising `http_retries` does not avoid this.
- **The block is not bounded.** Longest observed: 47590 s (13 h 13 m), still blocked when killed.

This is not gateway-specific. Any non-200 takes the same throw path, and S3 itself documents 5xx as expected transient responses that clients are required to retry.

There is a second consequence: the `CreateMultipartUpload` that succeeds during teardown creates an upload that is then never completed or aborted, so each occurrence also leaves an orphaned multipart upload on the server, billable until a lifecycle rule reaps it.

`uploads_in_progress` reads **1** in the debugger while blocked, on two independent runs — see below.

## Reproduction

Self-contained — no MinIO, no AWS, no object storage. The S3 endpoint is a stub inside the script and the injection is a single `if`.

```
pip install duckdb==1.5.5
python3 repro_hang.py # blocks; prints REPRODUCED after 60s, exits 0
FAIL_NTH=0 python3 repro_hang.py # control: no injection, COPY succeeds
```

The script lowers `s3_uploader_max_filesize` so that a ~14 MB output is enough to leave the single-buffer path. **The bug does not depend on that** — it reproduces on stock settings with an output larger than the default `part_size` (~76.5 MiB); the cap is lowered only to keep the test small and fast.

repro_hang.py

```python
#!/usr/bin/env python3
"""
Minimal self-contained reproduction: COPY TO 's3://...' blocks forever when one
CreateMultipartUpload (POST ?uploads=) fails and a later one succeeds.

Requires only: pip install duckdb==1.5.5
No MinIO / AWS / object storage needed - the S3 endpoint is a stub in this file.
Network is needed once, for pip and for INSTALL httpfs to fetch the extension;
after that the run talks only to the in-process stub on 127.0.0.1.

python3 repro_hang.py # blocks; prints REPRODUCED and exits 0
FAIL_NTH=0 python3 repro_hang.py # control: no injection, COPY succeeds

The stub is deliberately loud: it logs every request line *before* reading the
body, logs its own protocol errors, and times out idle sockets. That way "no
PUT ?partNumber= was ever sent" is something this script proves, rather than
something you have to take on faith - the stub cannot silently stall and
masquerade as a hang in DuckDB.
"""
import http.server
import os
import socketserver
import sys
import threading
import time
from urllib.parse import urlparse, parse_qs

BUCKET = "b"
KEY = "out.parquet"

# Which CreateMultipartUpload call to answer with 502. 1 = the first one
# (reproduces the hang). Set FAIL_NTH=0 to disable the injection entirely;
# the COPY then succeeds, which is the control case.
FAIL_NTH = int(os.environ.get("FAIL_NTH", "1"))

# How long to stay blocked before declaring the reproduction successful.
VERDICT_AFTER = int(os.environ.get("VERDICT_AFTER", "60"))

state_lock = threading.Lock()
create_mpu_calls = 0 # CreateMultipartUpload request lines seen
put_parts = 0 # PUT ?partNumber= request lines seen
reading_body = 0 # handlers currently blocked reading a request body
unexpected = 0 # requests the stub did not recognise

out_lock = threading.Lock()

def log(msg):
with out_lock:
sys.stdout.write(msg + "\n")
sys.stdout.flush()

class S3Stub(http.server.BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
# Without this an interrupted upload would park a handler in recv() forever,
# which looks exactly like the bug we are trying to demonstrate.
timeout = 30

def _respond(self, status, payload=b"", headers=None):
self.send_response(status)
for k, v in (headers or {}).items():
self.send_header(k, v)
if not (status == 204 and not payload):
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
if payload:
self.wfile.write(payload)

def _read_body(self):
"""Consume the request body exactly. Returns bytes actually read."""
if "chunked" in (self.headers.get("Transfer-Encoding") or "").lower():
# DuckDB sends Content-Length today. Rather than carry an untested
# chunked parser, fail loudly if that ever changes - a half-right
# parser would desync this keep-alive connection and the stub would
# start inventing failures of its own.
log(" !! UNEXPECTED: chunked request body; this stub only handles "
"Content-Length. Results from this run are not valid.")
with state_lock:
globals()["unexpected"] += 1
self._respond(501, b"chunked not supported by this stub")
return -1
declared = int(self.headers.get("Content-Length") or 0)
got = 0
with state_lock:
globals()["reading_body"] += 1
try:
while got < declared:
chunk = self.rfile.read(min(declared - got, 1 << 20))
if not chunk:
break
got += len(chunk)
finally:
with state_lock:
globals()["reading_body"] -= 1
if got != declared:
log(f" !! short read: declared {declared}, got {got}")
return got

# --- request classification -------------------------------------------

def _kind(self):
u = urlparse(self.path)
q = parse_qs(u.query, keep_blank_values=True)
if self.command == "POST" and "uploads" in q:
return "create_mpu"
if self.command == "POST" and "uploadId" in q:
return "complete_mpu"
if self.command == "PUT" and "partNumber" in q:
return "upload_part"
return None

def do_HEAD(self):
log(f" --> HEAD {self.path}")
self._respond(404)

def do_GET(self):
log(f" --> GET {self.path}")
self._respond(404)

def do_DELETE(self):
log(f" --> DELETE {self.path}")
self._respond(204)

def do_PUT(self):
# Log the request line BEFORE touching the body: if the body transfer
# ever stalls, the request must still show up in this log.
log(f" --> PUT {self.path}")
kind = self._kind()
if kind == "upload_part":
with state_lock:
globals()["put_parts"] += 1
n = self._read_body()
if n < 0:
return
if kind != "upload_part":
self._unexpected(n)
return
log(f" <-- 200 (part accepted, {n} bytes)")
self._respond(200, b"", {"ETag": '"stub-etag"'})

def do_POST(self):
log(f" --> POST {self.path}")
kind = self._kind()
n = self._read_body()
if n < 0:
return

if kind == "create_mpu":
with state_lock:
globals()["create_mpu_calls"] += 1
seq = create_mpu_calls
# >>> THE INJECTION: fail only the Nth CreateMultipartUpload <<<
if seq == FAIL_NTH:
log(f" <-- 502 (CreateMultipartUpload #{seq}) <-- injected")
self._respond(502, b"bad gateway", {"Content-Type": "text/plain"})
return
log(f" <-- 200 (CreateMultipartUpload #{seq})")
xml = (b''
b""
b"" + BUCKET.encode() + b""
b"" + KEY.encode() + b""
b"stub-upload-id"
b"")
self._respond(200, xml, {"Content-Type": "application/xml"})
return

if kind == "complete_mpu":
log(f" <-- 200 (CompleteMultipartUpload, {n} bytes)")
xml = (b''
b""
b"http://stub/" + BUCKET.encode() + b"/" + KEY.encode() + b""
b"" + BUCKET.encode() + b""
b"" + KEY.encode() + b""
b'"stub-etag"'
b"")
self._respond(200, xml, {"Content-Type": "application/xml"})
return

self._unexpected(n)

def _unexpected(self, n):
log(f" !! UNEXPECTED request the stub does not model: {self.command} "
f"{self.path} ({n} bytes). Results from this run are not valid.")
with state_lock:
globals()["unexpected"] += 1
self._respond(400, b"unexpected request")

# BaseHTTPRequestHandler routes its own protocol errors through log_message
# too (log_error calls it). Silencing everything would hide exactly the
# failures that would make this stub untrustworthy, so only drop the
# routine per-request access log.
def log_request(self, *a):
pass

def log_error(self, fmt, *a):
log(" !! stub protocol error: " + (fmt % a))

class Server(socketserver.ThreadingTCPServer):
allow_reuse_address = True
daemon_threads = True

def handle_error(self, request, client_address):
import traceback
log(" !! stub handler raised:\n" + traceback.format_exc())

def main():
httpd = Server(("127.0.0.1", 0), S3Stub)
port = httpd.server_address[1]
threading.Thread(target=httpd.serve_forever, daemon=True).start()
print(f"S3 stub listening on 127.0.0.1:{port}\n")

import duckdb
print("duckdb", duckdb.__version__)
# Persistent secrets in ~/.duckdb would take precedence over the settings
# below and silently send the requests somewhere else.
con = duckdb.connect(config={"allow_persistent_secrets": False})
con.execute("INSTALL httpfs; LOAD httpfs;")
print("httpfs", con.execute(
"SELECT extension_version FROM duckdb_extensions() "
"WHERE extension_name = 'httpfs'").fetchone()[0])
con.execute(f"SET s3_endpoint='127.0.0.1:{port}'")
con.execute("SET s3_use_ssl=false")
con.execute("SET s3_url_style='path'")
con.execute("SET s3_access_key_id='x'")
con.execute("SET s3_secret_access_key='y'")
con.execute("SET s3_region='us-east-1'")
con.execute("SET http_proxy=''")
# Default part_size is ~76.5 MiB, which would need a much larger output to
# leave the single-buffer path. Lowering the cap keeps the test small; the
# bug does not depend on it (it reproduces on defaults with a bigger file).
# part_size = ceil(max(5 MiB, max_filesize / max_parts_per_file) / BLK) * BLK
# with BLK = Storage::DEFAULT_BLOCK_SIZE = 262136 (262144 alloc - 8B header),
# so 50GB / 10000 -> 5,504,856 bytes (5.25 MiB), not 5 MiB.
con.execute("SET s3_uploader_max_filesize='50GB'")

print("\nrequest log:")
t0 = time.time()

def watchdog():
while True:
time.sleep(10)
el = int(time.time() - t0)
with state_lock:
c, p, r, u = create_mpu_calls, put_parts, reading_body, unexpected
# That this line keeps printing is itself evidence: the GIL is free,
# so the stub could serve any request that arrived. Nothing arrived.
log(f" ... blocked in COPY for {el}s "
f"(CreateMultipartUpload={c}, PUT parts={p}, "
f"handlers reading a body={r}, unexpected={u})")
if el >= VERDICT_AFTER and FAIL_NTH:
if u == 0 and r == 0 and p == 0 and c >= 2:
log(f"\nREPRODUCED: COPY has been blocked for {el}s. "
f"{c} CreateMultipartUpload requests, zero PUT parts, "
f"no handler stuck on a body, no unexpected requests — "
f"the uploader thread was never created and the stub is "
f"demonstrably idle and healthy.")
# The main thread is inside native code; sys.exit() from
# this thread would only end this thread.
os._exit(0)
log(f"\nINCONCLUSIVE after {el}s: expected 0 PUT parts / 0 stuck "
f"handlers / 0 unexpected and >=2 CreateMultipartUpload, got "
f"{p}/{r}/{u}/{c}. Do not treat this run as a reproduction.")
os._exit(2)

threading.Thread(target=watchdog, daemon=True).start()

try:
con.execute(
"COPY (SELECT i, (random()*1e18)::BIGINT a, (random()*1e18)::BIGINT b, "
" (random()*1e18)::BIGINT c, (random()*1e18)::BIGINT d "
" FROM range(400000) t(i)) "
f"TO 's3://{BUCKET}/{KEY}' (FORMAT PARQUET)")
ok = FAIL_NTH == 0
print(f"\nCOPY returned after {time.time() - t0:.1f}s"
f"{' -- expected, control case' if ok else ' -- NOT reproduced'}")
sys.exit(0 if ok else 1)
except Exception as e:
print(f"\nCOPY raised after {time.time() - t0:.1f}s: "
f"{type(e).__name__}: {e} -- NOT reproduced (expected a block)")
sys.exit(1)

if __name__ == "__main__":
main()
```

### Observed

```
duckdb 1.5.5
httpfs 827222f

request log:
--> POST /b/out.parquet?uploads=
<-- 502 (CreateMultipartUpload #1) <-- injected
--> HEAD /b/out.parquet
--> POST /b/out.parquet?uploads=
<-- 200 (CreateMultipartUpload #2)
... blocked in COPY for 10s (CreateMultipartUpload=2, PUT parts=0, handlers reading a body=0, unexpected=0)
... blocked in COPY for 20s (CreateMultipartUpload=2, PUT parts=0, handlers reading a body=0, unexpected=0)
...
REPRODUCED: COPY has been blocked for 60s. 2 CreateMultipartUpload requests, zero PUT
parts, no handler stuck on a body, no unexpected requests — the uploader thread was
never created and the stub is demonstrably idle and healthy.
```

Reading the three requests: #1 is `FlushBuffer` initialising the multipart upload and getting the injected 502. The `HEAD` is the failure cleanup path (`RemoveFile` → `OpenFile(..., NULL_IF_NOT_EXISTS)`). #2 is `FlushAllBuffers` initialising again during teardown; it succeeds, and the wait right after it never returns.

The stub logs every request line *before* reading the body, counts handlers blocked on a body read, times out idle sockets, and reports its own protocol errors — so "no PUT was sent" is something the script demonstrates rather than something you have to assume. The watchdog line continuing to print also shows the GIL is free: the stub could have served anything that arrived. Nothing arrived.

### The counter, read directly

The stock PyPI build has no DWARF, so the member cannot be printed by name — but it can be located from the kernel's own record of what the thread is waiting on, with no debug info required:

- `/proc/1/task/1/syscall` gives the futex address the thread is parked on. That word lives inside the `pthread_cond_t` of `final_flush_cv`, at offset 32 on glibc 2.41.
- The header declares `mutex uploads_in_progress_lock; condition_variable uploads_in_progress_cv; condition_variable final_flush_cv; uint16_t uploads_in_progress;` in that order, and `sizeof(std::mutex) == sizeof(std::condition_variable) == 48` in this image (measured, not assumed). So `uploads_in_progress` sits at `final_flush_cv + 48`, and `uploads_in_progress_lock` at `final_flush_cv - 96`.

```
futex (from kernel) = 0xaaab0dd732d8
final_flush_cv = 0xaaab0dd732b8
uploads_in_progress_lock first word = 0 # released by the wait, as expected
uploads_in_progress_cv first word = 0 # never used - the capacity wait was never hit
uploads_in_progress @ 0xaaab0dd732e8 = 1 # <<<
```

Two independent runs at different base addresses both read **1**. The all-zero `uploads_in_progress_cv` is a useful cross-check on the offsets: that condition variable is only touched when `max_upload_threads` is reached, which never happened here, so it should be pristine — and it is.

### Expected

`COPY` should raise the HTTP error, or complete. It should not block indefinitely.

### Control

```
--> POST /b/out.parquet?uploads=
<-- 200 (CreateMultipartUpload #1)
--> PUT /b/out.parquet?partNumber=1&uploadId=stub-upload-id
<-- 200 (part accepted, 5504856 bytes)
--> PUT /b/out.parquet?partNumber=3&uploadId=stub-upload-id
<-- 200 (part accepted, 3394945 bytes)
--> PUT /b/out.parquet?partNumber=2&uploadId=stub-upload-id
<-- 200 (part accepted, 5504856 bytes)
--> POST /b/out.parquet?uploadId=stub-upload-id
<-- 200 (CompleteMultipartUpload, 288 bytes)

COPY returned after 0.1s -- expected, control case
```

Same script, same data, one constant changed. (Exact byte counts vary a little per run — the payload uses `random()`.)

### Native stack while hung

Verbatim `eu-stack -p 1`, frames `#3`–`#21`, with only the leading unsymbolised frames removed:

```
#3 pthread_cond_wait
#4 std::condition_variable::wait(std::unique_lock&)
#5 duckdb::S3MultiPartUpload::FlushAllBuffers()
#6 duckdb::S3MultiPartUpload::Finalize()
#7 duckdb::S3FileHandle::~S3FileHandle()
#8 duckdb::S3FileHandle::~S3FileHandle()
#9 duckdb::BufferedFileWriter::~BufferedFileWriter()
#10 duckdb::ParquetWriter::~ParquetWriter()
#11 duckdb::ParquetWriteGlobalState::~ParquetWriteGlobalState()
#12 duckdb::CopyToFunctionGlobalState::~CopyToFunctionGlobalState()
#13 duckdb::CopyToFunctionGlobalState::~CopyToFunctionGlobalState()
#14 duckdb::PhysicalOperator::~PhysicalOperator()
#15 duckdb::PreparedStatementData::~PreparedStatementData()
#16 std::_Sp_counted_base<(__gnu_cxx::_Lock_policy)2>::_M_release()
#17 duckdb::ClientContext::EndQueryInternal(duckdb::ClientContextLock&, bool, bool, duckdb::optional_ptr)
#18 duckdb::ClientContext::ExecuteTaskInternal(duckdb::ClientContextLock&, duckdb::BaseQueryResult&, bool)
#19 duckdb::PendingQueryResult::ExecuteInternal(duckdb::ClientContextLock&)
#20 duckdb::PendingQueryResult::Execute()
#21 _duckdb_jdbc_execute_pending(JNIEnv_*, _jclass*, _jobject*)
```

**Provenance:** this stack, the 47590 s figure, and the `setQueryTimeout` result come from a second rig — the same scenario driven through the JDBC client against MinIO behind a fault-injecting proxy — because that is where the incident was first reproduced and where `eu-stack` was available. Everything else on this page comes from `repro_hang.py`. The two agree on the frames from `FlushAllBuffers` up to `PendingQueryResult`; only the client entry frame differs (`DuckDBPyConnection::CompletePendingQuery` on the Python side). While hung: zero TCP connections to the endpoint, zero traffic, ~0.2 % CPU, every thread in `futex_wait_queue`.

## Root cause

`src/s3_multi_part_upload.cpp` on 1.5.x (`src/s3/s3_multi_part_upload.cpp` on current `main`), `S3MultiPartUpload::FlushBuffer`:

```cpp
{
unique_lock lck(uploads_in_progress_lock);
#ifndef SAME_THREAD_UPLOAD
if (uploads_in_progress >= config_params.max_upload_threads) {
uploads_in_progress_cv.wait(lck, [&] { ... });
}
#endif
uploads_in_progress++; // (1) slot reserved
}
if (initialized_multipart_upload == false) {
multipart_upload_id = InitializeMultipartUpload(); // (2) throws on non-200
}
#ifdef SAME_THREAD_UPLOAD
UploadBuffer(shared_from_this(), write_buffer);
return;
#endif
std::thread upload_thread(S3MultiPartUpload::UploadBuffer, shared_from_this(), write_buffer);
upload_thread.detach(); // (3) unreachable if (2) throws
```

`InitializeMultipartUpload()` issues `POST ...?uploads=` and throws `HTTPException` on a non-200 status.

The only decrement is in `NotifyUploadsInProgress()`, whose only caller is `UploadBuffer` — i.e. the work started at (3). So when (2) throws, the slot reserved at (1) is never released:

```
$ grep -nE 'uploads_in_progress\b|NotifyUploadsInProgress' src/s3/s3_multi_part_upload.cpp
13: path(...), config_params(...), uploads_in_progress(0), parts_uploaded(0),
127: multi_part_upload->NotifyUploadsInProgress(); # only caller, from UploadBuffer
178:void S3MultiPartUpload::NotifyUploadsInProgress() {
181: if (uploads_in_progress == 0) {
183: "S3MultiPartUpload: uploads_in_progress decremented but no uploads are supposed to be active");
185: uploads_in_progress--; # only decrement
220: if (uploads_in_progress >= config_params.max_upload_threads) {
223: return uploads_in_progress < config_params.max_upload_threads;
227: uploads_in_progress++; # only increment
276: final_flush_cv.wait(lck, [&]() DUCKDB_REQUIRES(...) { return uploads_in_progress == 0; });
```

`FlushAllBuffers()` then waits on a predicate that can no longer become true:

```cpp
final_flush_cv.wait(lck, [&] { return uploads_in_progress == 0; });
```

This wait has no timeout. Note this is a leaked-state bug, not a lost wakeup — the wait is predicate-based, so an extra `notify` would not help.

**A failing part upload does *not* leak**, which is why only this one request matters: `UploadBufferImplementation` catches IO/HTTP errors, pushes them into the error manager and returns, and `UploadBuffer` still calls `NotifyUploadsInProgress()`. Only the window before the thread exists is unprotected.

**The `SAME_THREAD_UPLOAD` (Emscripten) build is not affected**: `UploadBuffer` runs inline there, and `final_flush_cv.wait` is compiled out entirely.

### Why the second request matters

`InitializeMultipartUpload()` is also called from `FlushAllBuffers()` during teardown. If that call fails too, teardown throws before reaching the wait and there is no hang. So the hang needs one `CreateMultipartUpload` to fail and a later one to succeed — which is why intermittent backend failures make this look nondeterministic. It does not require the backend to have fully recovered; one success is enough.

Two details that would otherwise look odd:

- Teardown re-initialises rather than taking `UploadSingleBuffer`, because `FlushBuffer` already erased the buffer from `write_buffers` before throwing, so `to_flush.size() == 1` no longer holds.
- Destruction reaches `Close()` at all even though the write path threw: `~S3FileHandle()` does have an `Exception::UncaughtException()` early return, but by the time `CopyToFunctionGlobalState` is destroyed the executor has already caught the task exception, so unwinding is not in progress. This is consistent with the stack above.

### Not reached on the single-buffer path

If the output is strictly smaller than `part_size`, teardown takes `UploadSingleBuffer` and never issues `POST ?uploads=` at all — verified: with the stub set to fail that request, it is never requested and the copy succeeds. That path additionally requires `kms_key_id` to be empty (`s3_multi_part_upload.cpp`).

## Why nothing recovers from it

- `http_timeout` does not apply — no HTTP request is in flight at that point.
- `http_retries` does not apply either, as above: the failing `CreateMultipartUpload` is not retried.
- Query cancellation does not apply — execution has finished; this is destructor teardown, past the interrupt checkpoints. A JDBC `setQueryTimeout(60)` did not fire after 460 s.
- The wait itself has no timeout, so the only way out is to kill the process.

## Versions

Reproduced on **1.5.4** and **1.5.5** (latest release at time of writing), Python and JDBC clients, httpfs extension `827222f`. Environment: `python:3.12-slim`, Debian 12, linux/arm64, CPython 3.12; JDBC run on `eclipse-temurin:21-jre`. Nothing looks platform-specific, but I have only run it on arm64 Linux.

Still present on `main` at `497adbd1` (2026-07-30). That commit touched this file — locks became `annotated_mutex` / `annotated_unique_lock` with `DUCKDB_REQUIRES` annotations, and an earlier commit moved the file under `src/s3/` — but the increment, the throwing call between it and the thread creation, and the untimed wait are all unchanged.

## Related, but as far as I can tell not duplicates

- **#304** — S3 write path, opposite symptom: errors swallowed, `COPY` returns success. Its repro writes a single row, which stays on the single-buffer path.
- **#347 / #335** — also a hang, different mechanism: self-deadlock re-entering the non-recursive `CachedFile::lock` on a region redirect, on the read path.
- **#324** — a per-request cancellation hook would not help: nothing is in flight at the point of the hang.
- **duckdb/duckdb#14877** — multipart upload that neither writes the file nor closes the handle, idling indefinitely; closed as stale without diagnosis. Parts were already uploaded there and none are here, so possibly unrelated; noting it only as something cheap to re-check afterwards.
- **duckdb/duckdb#9376** — compressed CSV `COPY TO` S3 never completing the multipart upload. Same family, much older code.

No open PR touches `FlushBuffer` as of `497adbd1`.

## Suggested fix

The invariant: the slot reserved by `uploads_in_progress++` should be released on **every** path that fails to hand it to an uploader, i.e. anywhere between the increment and the successful return of the `std::thread` constructor. Today the reproduced failure is `InitializeMultipartUpload()` throwing; `std::thread` construction throwing has the same shape and consequence.

One minimal option is a `catch`/rethrow around the initialization that calls `NotifyUploadsInProgress()` before rethrowing. That is safe here because the current thread has just performed the `++`, so the counter is `>= 1` and the zero-check inside `NotifyUploadsInProgress()` cannot fire — though that argument does rest on there being exactly one decrement site today. Simply moving the `{ lock; capacity wait; ++ }` block to *after* the initialization would also close the reproduced path without any exception handling.

If you prefer one guard covering both paths, two constraints are worth knowing:

- A scope guard's destructor is implicitly `noexcept`, so letting `NotifyUploadsInProgress()`'s `InternalException` escape it would call `std::terminate` — unconditionally, not only while unwinding. The guard would need to decrement directly, or swallow that throw.
- It has to be dismissed as soon as the `std::thread` constructor returns (in the `SAME_THREAD_UPLOAD` build, before the inline `UploadBuffer` call), otherwise the success path decrements twice.

Separately: `final_flush_cv.wait` having no timeout is what turns any such inconsistency into a process that can only be recovered by killing it. A bounded wait would be a trade-off rather than a free win — a legitimately slow upload of a large part over a narrow link must not be killed by it — so something like a wait that logs the counter and only gives up after a generous multiple of `http_timeout`, or an assertion in debug builds, may fit better than a hard deadline.

Run the self-contained repro_hang.py with the default FAIL_NTH and with FAIL_NTH=0 to confirm the hang and control case. Trace S3MultiPartUpload::FlushBuffer, uploads_in_progress, and S3FileHandle destruction, then verify that a failed CreateMultipartUpload no longer blocks teardown or leaves the later upload unfinished.

Written by the indexing model from the issue text.