Fix incomplete get handling via HTTP compat bump and don't write failed retry streams to parent buffer in _http_request

Project.toml

@@ -24,7 +24,7 @@ XMLDict = "228000da-037f-5747-90a9-8195ccbf91a5"
 [compat]
 Compat = "3.29"
 GitHub = "5"
-HTTP = "0.9.6"
+HTTP = "0.9.17"
 IniFile = "0.5"
 JSON = "0.18, 0.19, 0.20, 0.21"
 MbedTLS = "0.6, 0.7, 1"

src/utilities/downloads_backend.jl

@@ -65,7 +65,10 @@ function _http_request(backend::DownloadsBackend, request::Request, response_str
     # We pass an `input` only when we have content we wish to send.
     input = !isempty(request.content) ? IOBuffer(request.content) : nothing
 
-    @repeat 4 try
+    attempts = 4
+    attempt = 0
+    @repeat attempts try
+        attempt += 1
         downloader = @something(backend.downloader, get_downloader())
         # set the hook so that we don't follow redirects. Only
         # need to do this on per-request downloaders, because we
@@ -79,16 +82,38 @@ function _http_request(backend::DownloadsBackend, request::Request, response_str
         # but the first will send an empty payload.
         input !== nothing && seekstart(input)
 
-        response = @mock Downloads.request(
-            request.url;
-            input=input,
-            output=output,
-            method=request.request_method,
-            headers=request.headers,
-            verbose=false,
-            throw=true,
-            downloader=downloader,
-        )
+        # Because Downloads will write incomplete stream failures to the provided output buffer
+        # a sacrificial buffer is needed so that incomplete data can be discarded
+        buffer = isnothing(output) ? nothing : Base.BufferStream()
+        should_write = true
+        response = try
+            @mock Downloads.request(
+                request.url;
+                input=input,
+                output=buffer,
+                method=request.request_method,
+                headers=request.headers,
+                verbose=false,
+                throw=true,
+                downloader=downloader,
+            )
+        catch e
+            if e isa Downloads.RequestError && e.code == 18
+                # Downloads.RequestError 18 indicates an incomplete transfer, so don't pass the buffer forward
+                should_write = false
+            end
+            rethrow()
+        finally
+            if !isnothing(output)
+                close(buffer)
+                # Transfer the contents of the `BufferStream` into `response_stream` variable.
+                # but only if no EOFError error because of a broken connection OR it's the final attempt.
+                # i.e. Multiple EOFError retries shouldn't be passed to the `response_stream`
+                if should_write || attempt == attempts
+                    write(response_stream, buffer)
+                end
+            end
+        end
 
         http_response = HTTP.Response(
             response.status, response.headers; body=body_was_streamed, request=nothing

src/utilities/request.jl

@@ -144,7 +144,10 @@ end
 function _http_request(http_backend::HTTPBackend, request::Request, response_stream::IO)
     http_options = merge(http_backend.http_options, request.http_options)
 
-    @repeat 4 try
+    attempts = 4
+    attempt = 0
+    @repeat attempts try
+        attempt += 1
         # HTTP options such as `status_exception` need to be used when creating the stack
         http_stack = HTTP.stack(;
             redirect=false, retry=false, aws_authorization=false, http_options...
@@ -156,7 +159,7 @@ function _http_request(http_backend::HTTPBackend, request::Request, response_str
         # will keep using the `response_stream` kwarg to ensure we aren't relying on the
         # response's body being populated.
         buffer = Base.BufferStream()
-
+        should_write = true
         r = try
             @mock HTTP.request(
                 http_stack,
@@ -168,14 +171,24 @@ function _http_request(http_backend::HTTPBackend, request::Request, response_str
                 response_stream=buffer,
                 http_options...,
             )
+        catch e
+            if e isa HTTP.IOError && e.e isa EOFError
+                # EOFErrors indicate a broken connection, so don't pass the buffer forward
+                should_write = false
+            end
+            rethrow()
         finally
             # We're unable to read from the `Base.BufferStream` until it has been closed.
             # HTTP.jl will close passed in `response_stream` keyword. This ensures that it
             # is always closed (e.g. HTTP.jl 0.9.15)
             close(buffer)
 
-            # Transfer the contents of the `BufferStream` into `response_stream` variable.
-            write(response_stream, buffer)
+            # Transfer the contents of the `BufferStream` into `response_stream` variable
+            # but only if no EOFError error because of a broken connection OR it's the final attempt.
+            # i.e. Multiple EOFError retries shouldn't be passed to the `response_stream`
+            if should_write || attempt == attempts
+                write(response_stream, buffer)
+            end
         end
 
         return @mock Response(r, response_stream)

test/issues.jl

@@ -155,6 +155,79 @@ try
         end
     end
 
+    @testset "issue 515" begin
+        # https://github.com/JuliaCloud/AWS.jl/issues/515
+        n = 100
+        n_incomplete = n - 1
+        data = rand(UInt8, n)
+
+        @testset "Fail 2 attempts then succeed" begin
+            i = 2
+            ATTEMPT_NUM = 0 # reset
+            patch = if AWS.DEFAULT_BACKEND[] isa AWS.HTTPBackend
+                @patch function HTTP.request(args...; response_stream, kwargs...)
+                    ATTEMPT_NUM += 1
+                    if ATTEMPT_NUM <= i
+                        write(response_stream, data[1:n_incomplete]) # an incomplete stream that shouldn't be retained
+                        throw(HTTP.IOError(EOFError(), "msg"))
+                    else
+                        write(response_stream, data)
+                        return HTTP.Response(200, "{\"Location\": \"us-east-1\"}")
+                    end
+                end
+            elseif AWS.DEFAULT_BACKEND[] isa AWS.DownloadsBackend
+                @patch function Downloads.request(args...; output, kwargs...)
+                    ATTEMPT_NUM += 1
+                    if ATTEMPT_NUM <= i
+                        write(output, data[1:n_incomplete]) # an incomplete stream that shouldn't be retained
+                        throw(Downloads.RequestError("", 18, "transfer closed with 1 bytes remaining to read", Downloads.Response("http", "", 200, "HTTP/1.1 200 OK", ["content-length" => "64"])))
+                    else
+                        write(output, data)
+                        return Downloads.Response("http", "", 200, "HTTP/1.1 200 OK", ["content-length" => "64"])
+                    end
+                end
+            end
+            config = AWSConfig(; creds=nothing)
+            apply(patch) do
+                resp = S3.get_object("www.invenia.ca", "index.html"; aws_config=config) # use public bucket as dummy
+                @test length(resp) == n
+                @test resp == data
+            end
+        end
+        @testset "Fail all 4 attempts then throw" begin
+            i = 4
+            ATTEMPT_NUM = 0 # reset
+            patch = if AWS.DEFAULT_BACKEND[] isa AWS.HTTPBackend
+                @patch function HTTP.request(args...; response_stream, kwargs...)
+                    ATTEMPT_NUM += 1
+                    if ATTEMPT_NUM <= i
+                        write(response_stream, data[1:n_incomplete]) # an incomplete stream that shouldn't be retained
+                        throw(HTTP.IOError(EOFError(), "msg"))
+                    else
+                        write(response_stream, data)
+                        return HTTP.Response(200, "{\"Location\": \"us-east-1\"}")
+                    end
+                end
+            elseif AWS.DEFAULT_BACKEND[] isa AWS.DownloadsBackend
+                @patch function Downloads.request(args...; output, kwargs...)
+                    ATTEMPT_NUM += 1
+                    if ATTEMPT_NUM <= i
+                        write(output, data[1:n_incomplete]) # an incomplete stream that shouldn't be retained
+                        throw(Downloads.RequestError("", 18, "transfer closed with 1 bytes remaining to read", Downloads.Response("http", "", 200, "HTTP/1.1 200 OK", ["content-length" => "64"])))
+                    else
+                        write(output, data)
+                        return Downloads.Response("http", "", 200, "HTTP/1.1 200 OK", ["content-length" => "64"])
+                    end
+                end
+            end
+            config = AWSConfig(; creds=nothing)
+            apply(patch) do
+                err_t = AWS.DEFAULT_BACKEND[] isa AWS.HTTPBackend ? HTTP.IOError : Downloads.RequestError
+                @test_throws err_t S3.get_object("www.invenia.ca", "index.html"; aws_config=config) # use public bucket as dumm
+            end
+        end
+    end
+
 finally
     S3.delete_bucket(BUCKET_NAME)
 end

test/runtests.jl

@@ -20,6 +20,7 @@ using AWS.AWSMetadata:
 using Base64
 using Compat: mergewith
 using Dates
+using Downloads
 using GitHub
 using HTTP
 using IniFile: Inifile