[fix](s3) Keep the response stream usable when an error body overflows the read buffer

be/src/io/fs/err_utils.cpp

@@ -124,19 +124,22 @@ Status localfs_error(int posix_errno, std::string_view msg) {
 
 Status s3fs_error(const Aws::S3::S3Error& err, std::string_view msg) {
     using namespace Aws::Http;
+    // A failure raised by the client itself carries no request id. Printing nothing leaves a
+    // dangling `request_id=` that has been read as a request id of the object storage.
+    std::string request_id = err.GetRequestId().empty() ? "<empty>" : err.GetRequestId().c_str();
     switch (err.GetResponseCode()) {
     case HttpResponseCode::NOT_FOUND:
         return Status::Error<NOT_FOUND, false>("{}: {} {} code=NOT_FOUND, type={}, request_id={}",
                                                msg, err.GetExceptionName(), err.GetMessage(),
-                                               err.GetErrorType(), err.GetRequestId());
+                                               err.GetErrorType(), request_id);
     case HttpResponseCode::FORBIDDEN:
         return Status::Error<PERMISSION_DENIED, false>(
                 "{}: {} {} code=FORBIDDEN, type={}, request_id={}", msg, err.GetExceptionName(),
-                err.GetMessage(), err.GetErrorType(), err.GetRequestId());
+                err.GetMessage(), err.GetErrorType(), request_id);
     default:
         return Status::Error<ErrorCode::INTERNAL_ERROR, false>(
                 "{}: {} {} code={} type={}, request_id={}", msg, err.GetExceptionName(),
-                err.GetMessage(), err.GetResponseCode(), err.GetErrorType(), err.GetRequestId());
+                err.GetMessage(), err.GetResponseCode(), err.GetErrorType(), request_id);
     }
 }
 

be/src/io/fs/s3_common.h

@@ -20,6 +20,11 @@
 #include <aws/core/utils/memory/stl/AWSStreamFwd.h>
 #include <aws/core/utils/stream/PreallocatedStreamBuf.h>
 
+#include <algorithm>
+#include <cstring>
+#include <streambuf>
+#include <vector>
+
 namespace doris {
 
 // A non-copying iostream.
@@ -34,12 +39,132 @@ class StringViewStream : Aws::Utils::Stream::PreallocatedStreamBuf, public std::
               std::iostream(this) {}
 };
 
+// The AWS SDK writes the body of every response into the stream built by the response
+// stream factory of the request, whatever the status of that response is. Reading an
+// object range straight into the buffer of the caller therefore breaks as soon as the
+// server answers with an error: the XML body of a `429 SlowDown` is a few hundred bytes
+// and does not fit into the buffer of a small range read. `PreallocatedStreamBuf` does not
+// implement `overflow()`, so the stream turns bad, curl aborts the transfer with
+// `CURLE_WRITE_ERROR`, and the SDK reports an `INTERNAL_FAILURE` named "Failed to flush
+// response stream" while never recording the status code of the response. Both the retry
+// strategy of the SDK and the retry of `S3FileReader` key on that status code, so an error
+// the server asked us to retry ends up cancelling the query instead.
+//
+// This stream buffer writes into the buffer of the caller as long as the body fits, which
+// is the case for every successful ranged read, and spills the rest into a buffer of its
+// own. The stream never turns bad, so the SDK reports the real status code and can parse
+// the error out of the body.
+class ResponseStreamBuf final : public std::streambuf {
+public:
+    // Bodies beyond this size are truncated. Only error documents are expected to overflow
+    // and their leading bytes already carry the error code and the message.
+    static constexpr size_t MAX_SPILL_SIZE = 1024 * 1024;
+
+    ResponseStreamBuf(void* buf, size_t nbytes) : _buf(static_cast<char*>(buf)) {
+        setp(_buf, _buf + nbytes);
+        setg(_buf, _buf, _buf);
+    }
+
+protected:
+    std::streamsize xsputn(const char* s, std::streamsize n) override {
+        if (!_spilled) {
+            if (n <= epptr() - pptr()) {
+                std::memcpy(pptr(), s, n);
+                pbump(static_cast<int>(n));
+                return n;
+            }
+            _spill_over();
+        }
+        auto writable = std::min(static_cast<size_t>(n), MAX_SPILL_SIZE - _spill.size());
+        _spill.insert(_spill.end(), s, s + writable);
+        // Always report the whole write as consumed. A short write is what makes curl
+        // abort the transfer and lose the status code of the response.
+        return n;
+    }
+
+    int_type overflow(int_type ch) override {
+        if (traits_type::eq_int_type(ch, traits_type::eof())) {
+            return traits_type::not_eof(ch);
+        }
+        auto c = traits_type::to_char_type(ch);
+        xsputn(&c, 1);
+        return ch;
+    }
+
+    int_type underflow() override {
+        _reset_get_area(_read_pos());
+        if (gptr() == egptr()) {
+            return traits_type::eof();
+        }
+        return traits_type::to_int_type(*gptr());
+    }
+
+    pos_type seekoff(off_type off, std::ios_base::seekdir dir,
+                     std::ios_base::openmode which) override {
+        auto size = static_cast<off_type>(_written());
+        if (which & std::ios_base::out) {
+            // The SDK only asks for the write position, to tell an empty body apart from a
+            // body it has to parse. Moving the write pointer is not supported.
+            return dir == std::ios_base::cur && off == 0 ? pos_type(size) : pos_type(off_type(-1));
+        }
+        off_type pos = off;
+        if (dir == std::ios_base::cur) {
+            pos += static_cast<off_type>(_read_pos());
+        } else if (dir == std::ios_base::end) {
+            pos += size;
+        }
+        if (pos < 0 || pos > size) {
+            return pos_type(off_type(-1));
+        }
+        _reset_get_area(static_cast<size_t>(pos));
+        return pos_type(pos);
+    }
+
+    pos_type seekpos(pos_type pos, std::ios_base::openmode which) override {
+        return seekoff(pos, std::ios_base::beg, which);
+    }
+
+private:
+    // Moves what has been written so far into the spill buffer, so that the body stays
+    // contiguous and the SDK can parse the error out of it.
+    void _spill_over() {
+        _spill.assign(_buf, pptr());
+        setp(nullptr, nullptr);
+        _spilled = true;
+    }
+
+    // Bytes of the body held by this buffer, truncation excluded.
+    size_t _written() const { return _spilled ? _spill.size() : pptr() - _buf; }
+
+    // Both areas start at the same logical offset, so the read position survives a spill.
+    size_t _read_pos() const { return gptr() - eback(); }
+
+    void _reset_get_area(size_t pos) {
+        char* begin = _spilled ? _spill.data() : _buf;
+        auto size = _written();
+        pos = std::min(pos, size);
+        setg(begin, begin + pos, begin + size);
+    }
+
+    char* _buf;
+    std::vector<char> _spill;
+    bool _spilled = false;
+};
+
+class ResponseStream final : public std::iostream {
+public:
+    ResponseStream(void* buf, size_t nbytes) : std::iostream(&_buf), _buf(buf, nbytes) {}
+
+private:
+    ResponseStreamBuf _buf;
+};
+
 // By default, the AWS SDK reads object data into an auto-growing StringStream.
 // To avoid copies, read directly into our preallocated buffer instead.
 // See https://github.com/aws/aws-sdk-cpp/issues/64 for an alternative but
 // functionally similar recipe.
 inline Aws::IOStreamFactory AwsWriteableStreamFactory(void* buf, int64_t nbytes) {
-    return [=]() { return Aws::New<StringViewStream>("", buf, nbytes); };
+    return [=]() { return Aws::New<ResponseStream>("", buf, static_cast<size_t>(nbytes)); };
 }
 
 } // namespace doris

be/src/io/fs/s3_file_reader.cpp

@@ -183,9 +183,10 @@ Status S3FileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_rea
                 total_sleep_time += wait_time;
                 continue;
             } else {
-                // Handle other errors
-                return std::move(Status(resp.status.code, std::move(resp.status.msg))
-                                         .append("failed to read"));
+                // Handle other errors. The message already tells what failed to be read,
+                // appending to it only leaves a trailing token behind the request id of the
+                // object storage, which has been read as a request id more than once.
+                return {resp.status.code, std::move(resp.status.msg)};
             }
         }
         if (*bytes_read != bytes_req) {

be/src/io/fs/s3_obj_storage_client.cpp

@@ -310,17 +310,22 @@ ObjectStorageResponse S3ObjStorageClient::get_object(const ObjectStoragePathOpti
     if (!outcome.IsSuccess()) {
         record_s3_request_failed(outcome.GetError());
         return {convert_to_obj_response(s3fs_error(
-                        outcome.GetError(), fmt::format("failed to read from {}", opts.key))),
+                        outcome.GetError(),
+                        fmt::format("failed to read from bucket={} key={} offset={} size={}",
+                                    opts.bucket, opts.key, offset, bytes_read))),
                 static_cast<int>(outcome.GetError().GetResponseCode()),
                 outcome.GetError().GetRequestId()};
     }
     *size_return = outcome.GetResult().GetContentLength();
-    // case for incomplete read
+    // case for incomplete read, and the case of a server or a proxy answering a ranged read
+    // with the whole object, which no longer fits into the buffer of the caller
     SYNC_POINT_CALLBACK("s3_obj_storage_client::get_object", size_return);
     if (*size_return != bytes_read) {
         return {convert_to_obj_response(Status::InternalError(
-                "failed to read from {}(bytes read: {}, bytes req: {}), request_id: {}", opts.key,
-                *size_return, bytes_read, outcome.GetResult().GetRequestId()))};
+                "failed to read from bucket={} key={} offset={}(bytes read: {}, bytes req: {}), "
+                "request_id: {}",
+                opts.bucket, opts.key, offset, *size_return, bytes_read,
+                outcome.GetResult().GetRequestId()))};
     }
     return ObjectStorageResponse::OK();
 }

be/test/io/fs/s3_response_stream_test.cpp

@@ -0,0 +1,151 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <sstream>
+#include <string>
+#include <vector>
+
+#include "io/fs/s3_common.h"
+
+namespace doris {
+
+namespace {
+
+// What the SDK does with the body of a response it has to build an error from.
+std::string drain(std::iostream& stream) {
+    std::stringstream out;
+    out << stream.rdbuf();
+    return out.str();
+}
+
+// The XML body a MinIO answers a throttled ranged read with, shortened.
+constexpr char SLOW_DOWN_BODY[] =
+        R"(<?xml version="1.0" encoding="UTF-8"?><Error><Code>SlowDown</Code><Message>Please )"
+        R"(reduce your request rate.</Message><Key>data/packed_file/2666/x.bin</Key></Error>)";
+
+} // namespace
+
+// A body of the requested size lands in the buffer of the caller, without a copy.
+TEST(ResponseStreamTest, BodyFits) {
+    std::string body(64, 'a');
+    std::vector<char> buffer(body.size());
+
+    ResponseStream stream(buffer.data(), buffer.size());
+    stream.write(body.data(), body.size());
+    stream.flush();
+
+    EXPECT_FALSE(stream.fail());
+    EXPECT_EQ(body, std::string(buffer.data(), buffer.size()));
+    EXPECT_EQ(static_cast<std::streampos>(body.size()), stream.tellp());
+    EXPECT_EQ(body, drain(stream));
+}
+
+// An error body larger than the range of the read leaves the stream usable, which is what
+// keeps curl from aborting the transfer and the SDK from losing the status code.
+TEST(ResponseStreamTest, ErrorBodyOverflowsInOneWrite) {
+    std::string body(SLOW_DOWN_BODY);
+    // A read of the footer of a packed file is far smaller than the error document.
+    std::vector<char> buffer(12);
+
+    ResponseStream stream(buffer.data(), buffer.size());
+    stream.write(body.data(), body.size());
+    stream.flush();
+
+    EXPECT_FALSE(stream.fail());
+    EXPECT_EQ(static_cast<std::streampos>(body.size()), stream.tellp());
+    EXPECT_EQ(body, drain(stream));
+}
+
+// curl hands the body over in chunks, so the overflow can happen in the middle of one.
+TEST(ResponseStreamTest, ErrorBodyOverflowsAcrossWrites) {
+    std::string body(SLOW_DOWN_BODY);
+    std::vector<char> buffer(16);
+
+    ResponseStream stream(buffer.data(), buffer.size());
+    size_t chunk = 7;
+    for (size_t pos = 0; pos < body.size(); pos += chunk) {
+        stream.write(body.data() + pos, std::min(chunk, body.size() - pos));
+    }
+    stream.flush();
+
+    EXPECT_FALSE(stream.fail());
+    EXPECT_EQ(static_cast<std::streampos>(body.size()), stream.tellp());
+    // The bytes written before the overflow are kept, so the body stays contiguous.
+    EXPECT_EQ(body, drain(stream));
+}
+
+// A body written one character at a time goes through overflow() instead of xsputn().
+TEST(ResponseStreamTest, ErrorBodyOverflowsCharByChar) {
+    std::string body(SLOW_DOWN_BODY);
+    std::vector<char> buffer(4);
+
+    ResponseStream stream(buffer.data(), buffer.size());
+    for (char c : body) {
+        stream.put(c);
+    }
+    stream.flush();
+
+    EXPECT_FALSE(stream.fail());
+    EXPECT_EQ(body, drain(stream));
+}
+
+// A server answering a ranged read with the whole object must not blow up the memory of the
+// backend. The body is truncated, the stream stays good and the read is rejected later on by
+// the length check of the caller.
+TEST(ResponseStreamTest, OversizedBodyIsTruncated) {
+    std::string body(ResponseStreamBuf::MAX_SPILL_SIZE + 4096, 'x');
+    std::vector<char> buffer(8);
+
+    ResponseStream stream(buffer.data(), buffer.size());
+    stream.write(body.data(), body.size());
+    stream.flush();
+
+    EXPECT_FALSE(stream.fail());
+    EXPECT_EQ(static_cast<std::streampos>(ResponseStreamBuf::MAX_SPILL_SIZE), stream.tellp());
+    EXPECT_EQ(ResponseStreamBuf::MAX_SPILL_SIZE, drain(stream).size());
+}
+
+// The SDK rewinds the body before parsing an error out of it.
+TEST(ResponseStreamTest, SeekBackAndForth) {
+    std::string body(SLOW_DOWN_BODY);
+    std::vector<char> buffer(12);
+
+    ResponseStream stream(buffer.data(), buffer.size());
+    stream.write(body.data(), body.size());
+
+    EXPECT_EQ(body, drain(stream));
+    stream.clear();
+    stream.seekg(0);
+    EXPECT_EQ(body, drain(stream));
+
+    stream.clear();
+    stream.seekg(2);
+    EXPECT_EQ(body.substr(2), drain(stream));
+}
+
+// An empty body is what tells the SDK to build the error out of the status code alone.
+TEST(ResponseStreamTest, EmptyBody) {
+    std::vector<char> buffer(16);
+    ResponseStream stream(buffer.data(), buffer.size());
+
+    EXPECT_EQ(std::streampos(0), stream.tellp());
+    EXPECT_TRUE(drain(stream).empty());
+}
+
+} // namespace doris