Error ioOnClosedChannel

[gulivero1773]

When I use _ = try await writer.eventLoop.flatSubmit { writer.write(.buffer(buffer)) }.get() I have this error Download file fail with ioOnClosedChannel, localizedDescription The operation couldn’t be completed. (NIOCore.ChannelError error 5.)

[RaghavRoy145]

Hello! I'm trying to reproduce this issue. It seems that the writer channel is already closed. (ioOnClosedChannel)

Can you share a minimal reproducible example or steps to trigger the error?

As a first step I'd want to add a simple check like:

guard writer.isActive else {
    print("Channel is not active.")
    return
}

[gulivero1773]

var headers: HTTPHeaders = [:] headers.contentType = .binary headers.contentLength = "\(byteCount)" req.logger.info("SUCCESS: Before init Response byteCount equal - \(byteCount)") let response = Response(status: .ok, headers: headers) response.body = .init( stream: { writer in _ = req.eventLoop.makeFutureWithTask { do { try await req.attachments.read(by: attachment.fileId) { buffer in response.headers.add(name: .acceptRanges, value: "bytes") response.status = .partialContent response.headers.add(name: .contentRange, value: "bytes \(buffer.slice())") req.logger.info("SUCCESS: Read buffer equal - \(buffer)") do { _ = try await writer.eventLoop.flatSubmit { writer.write(.buffer(buffer)) }.get() } catch { req.logger.error("ERROR:-------------- Download file fail with \(error), localizedDescription \(error.localizedDescription)") } } _ = try await writer.eventLoop.flatSubmit { writer.write(.end) }.get() } catch { req.logger.error("ERROR: Download file fail with \(error), localizedDescription \(error.localizedDescription)") _ = try await writer.eventLoop.flatSubmit { writer.write(.error(error)) }.get() } } }, count: byteCount, byteBufferAllocator: req.byteBufferAllocator )

[gulivero1773]

writer has no isActive

[RaghavRoy145]

Thanks for the full code! I added some comments that may point to some bugs.

response.body = .init(
    stream: { writer in
        _ = req.eventLoop.makeFutureWithTask {
            do {
                // You're starting a task here, but not synchronizing writes
                try await req.attachments.read(by: attachment.fileId) { buffer in
                    
                    response.headers.add(name: .acceptRanges, value: "bytes")
                    response.status = .partialContent
                    response.headers.add(name: .contentRange, value: "bytes \(buffer.slice())")
                    
                    req.logger.info("SUCCESS: Read buffer equal - \(buffer)")

                    // If the writer's channel is already closed by the time this executes, you'll get ioOnClosedChannel
                    do {
                        _ = try await writer.eventLoop.flatSubmit {
                            writer.write(.buffer(buffer)) // <-- Here
                        }.get()
                    } catch {
                        req.logger.error("ERROR:-------------- Download file fail with \(error), localizedDescription \(error.localizedDescription)")
                    }
                }

                _ = try await writer.eventLoop.flatSubmit {
                    writer.write(.end)
                }.get()
            } catch {
                req.logger.error("ERROR: Download file fail with \(error), localizedDescription \(error.localizedDescription)")

                _ = try await writer.eventLoop.flatSubmit {
                    writer.write(.error(error))
                }.get()
            }
        }
    },
    count: byteCount,
    byteBufferAllocator: req.byteBufferAllocator
)

To try reproducing it I created a small Vapor app with the issue and a potential fix, you can follow the steps to reproduce it at your end (if you have vapor installed) or try out a similar fix directly in your code:

vapor new BugTest --template=api
cd BugTest
base64 /dev/urandom | head -c 5MB > test.txt

replace routes.swift with:

import Vapor

func routes(_ app: Application) throws {
    app.get("download") { req -> Response in
        let fileIO = req.application.fileio
        let path = req.application.directory.workingDirectory + "test.txt"

        var headers = HTTPHeaders()
        headers.add(name: .contentType, value: "application/octet-stream")

        let response = Response(status: .ok, headers: headers)

        response.body = .init(stream: { writer in
            _ = req.eventLoop.makeFutureWithTask {
                do {
                    // Open file and get size
                    let handle = try await fileIO.openFile(path: path, mode: .read, eventLoop: req.eventLoop).get()
                    let attributes = try FileManager.default.attributesOfItem(atPath: path)
                    let fileSize = attributes[.size] as! Int

                    let chunkSize = 1024
                    var offset = 0

                    // `.read` with closure for each chunk
                    while offset < fileSize {
                        let end = min(offset + chunkSize, fileSize)

                        let buffer = try await fileIO.read(
                            fileRegion: .init(fileHandle: handle, readerIndex: offset, endIndex: end),
                            allocator: req.byteBufferAllocator,
                            eventLoop: req.eventLoop
                        ).get()

                        // unsafe usage
                        do {
                            _ = try await writer.eventLoop.flatSubmit {
                                writer.write(.buffer(buffer))
                            }.get()
                        } catch {
                            req.logger.error("ERROR writing buffer: \(error)")
                        }

                        offset = end
                    }

                    // premature .end write
                    _ = try await writer.eventLoop.flatSubmit {
                        writer.write(.end)
                    }.get()

                    try handle.close()
                } catch {
                    req.logger.error("ERROR streaming: \(error)")
                }
            }
        })

        return response
    }
}

Now if you swift run and then curl http://localhost:8080/download --limit-rate 1k in a different terminal, then you should see the same ioOnClosedChannel error eventually.

I had to make a large txt file and run the curl this way to simulate a slow client, running this locally was too fast to force a race otherwise.

For the fixed version, you can replace the old routes.swift code in your vapor app with:

import Vapor

func routes(_ app: Application) throws {
    app.get("download") { req -> Response in
        let fileIO = req.application.fileio
        let path = req.application.directory.workingDirectory + "test.txt"

        var headers = HTTPHeaders()
        headers.add(name: .contentType, value: "application/octet-stream")

        let response = Response(status: .ok, headers: headers)

        response.body = .init(stream: { writer in
            _ = req.eventLoop.makeFutureWithTask {
                do {
                    // Open file and get size
                    let handle = try await fileIO.openFile(path: path, mode: .read, eventLoop: req.eventLoop).get()
                    let attributes = try FileManager.default.attributesOfItem(atPath: path)
                    let fileSize = attributes[.size] as! Int

                    let chunkSize = 1024
                    var offset = 0

                    // Write chunks in sequence
                    while offset < fileSize {
                        let end = min(offset + chunkSize, fileSize)

                        let buffer = try await fileIO.read(
                            fileRegion: .init(fileHandle: handle, readerIndex: offset, endIndex: end),
                            allocator: req.byteBufferAllocator,
                            eventLoop: req.eventLoop
                        ).get()

                        writer.write(.buffer(buffer))  // no .flatSubmit or .get()
                        offset = end
                    }

                    writer.write(.end)  // safe
                    try handle.close()
                } catch {
                    req.logger.error("ERROR streaming: \(error)")
                }
            }
        })

        return response
    }
}

^ I've basically just avoided .get() to avoid blocking and no .flatSubmit {}, and also just using writer.write() directly.

I should add that I'm not sure if this is recommended, but it works for me.

[gulivero1773]

Thank you I will try it! If use local storage it work, but I use mongo DB and if I use this it not work. Maybe I lost connection to DB?

[RaghavRoy145]

Yeah the could be the case, I'd need to have more info if its exactly that, though it's worth experimenting with the fix. The Vapor example was just a way to reproduce this. In your actual app, if you're streaming data from MongoDB and writing it to a channel, the same principle applies: avoid blocking on futures and ensure your writes are properly sequenced. If the MongoDB connection drops mid-stream (say, due to a timeout or network issue), you should catch that error by either writing a final .error() (if the channel is still open) or skipping the write. In the Vapor app, I used a large file and curl --limit-rate to artificially slow down the stream and simulate the kind of dropped connection you might experience with MongoDB. Without this slowdown, the race condition often doesn’t occur for me just like in your localDB test where everything was fast and stable.