Abort chunked transfer before write_eof

[pedro-psb]

Hello!
I have a request handler in my server that writes a chunked response to the client.

I want to be able to abort the underlying connection in case I detect I've sent some wrong data, so the client on the other side has a clue something went wrong.

I'm testing this with curl and noticed even if I don't call response.write_eof(), curl receives one:

<= Recv header, 2 bytes (0x2)
0000:
<= Recv data, 27 bytes (0x1b)
0000: 10
0004: bbb.bbb.bbb.bbb.
0016: 0
0019:
== Info: Connection #0 to host localhost left intact

What is the appropriate way achieve that?

[Dreamsorcerer]

What is the code you have already? I suspect it writes the EOF when you successfully return from the handler. You can probably use something like request.transport.close() or something similar to abort the connection.

[pedro-psb]

Thanks for the quick response!
I still get the eof response 🤔

I'll look into that and try to come up with a minimum reproduction decoupled from our specific app.

Here is a simplified version of how the code looks like. Runs with gunicorn.

app = web.Application(middlewares=[authenticate])
app.add_routes([web.get(path_prefix + "{path:.+}", Handler().stream_content)])


class Handler:
    async def stream_content(self, request):
        # ...
        # then:
        try:
            return await self.stream_remote_artifact(
                request,
                StreamResponse(headers=headers),
                remote_artifact,  # contains url, expected checksum, etc of remote artifact
            )
        except ClientResponseError as ce:
            raise Error(...)

    async def stream_remote_artifact(self, request, response, remote_artifact):
        # in there we handle headers, call prepare and write data as we receive it
        downloader = remote.get_downloader(remote_artifact, response)

        try:
            download_result = await downloader.run()
        except DigestValidationError:
            # That's what I'm trying, no visible effect
            request.transport.close()
            # can force 'ConnectionResetError: Cannot write to closing transport' with response.write(b'a')
            # but also doesnt help
            return response

        await response.write_eof()
        return response

[Dreamsorcerer]

request.transport.abort()? There's a few things you can look at the request, maybe one of them will do it..

[pedro-psb]

Yeah, I tried a coulple of those. I went as far as doing request.transport._sock.shutdown(socket.RDWR) / close().
Will keep looking, maybe I'm missing some detail. Thanks for the feedback!

[pedro-psb]

Both request.transport.close() and abort() works on a minimum example.
There something in the app code interfering with that.

from aiohttp import web

async def hello(request):
    headers = {
        "Transfer-Encoding": "chunked",
    }
    response = web.StreamResponse(headers=headers)
    await response.prepare(request)
    await response.write(b"foo")
    await response.write(b"spam")
    request.transport.close()
    return response

app = web.Application()
app.add_routes([web.get("/", hello)])
web.run_app(app)

<= Recv data, 8 bytes (0x8)
0000: 3
0003: foo
<= Recv data, 9 bytes (0x9)
0000: 4
0003: spam
<= Recv data, 0 bytes (0x0)
== Info: transfer closed with outstanding read data remaining
== Info: Closing connection

[pedro-psb]

I suspect there is something about the gunicorn deployment that interferes with that.

Anyway, I was able to make it work with the following:

close_tcp_connection(request.transport._sock)

(...)

def close_tcp_connection(sock):
    """Configure socket to close TCP connection immediately."""
    try:
        l_onoff = 1
        l_linger = 0  # 0 seconds timeout - immediate close
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", l_onoff, l_linger))
    except (socket.error, OSError) as e:
        log.warning(f"Error configuring socket for force close: {e}")