fix(channel): handle writing to file instead of pipe (#30519)

src/nvim/event/defs.h

@@ -86,8 +86,9 @@ struct stream {
     uv_tty_t tty;
 #endif
   } uv;
-  uv_stream_t *uvstream;
+  uv_stream_t *uvstream;  ///< NULL when the stream is a file
-  uv_file fd;
+  uv_file fd;    ///< When the stream is a file, this is its file descriptor
+  int64_t fpos;  ///< When the stream is a file, this is the position in file
   void *cb_data;
   stream_close_cb close_cb, internal_close_cb;
   void *close_cb_data, *internal_data;
@@ -112,7 +113,6 @@ struct rstream {
   uv_buf_t uvbuf;
   stream_read_cb read_cb;
   size_t num_bytes;
-  int64_t fpos;
 };
 
 #define ADDRESS_MAX_SIZE 256

src/nvim/event/rstream.c

@@ -34,7 +34,6 @@ void rstream_init_stream(RStream *stream, uv_stream_t *uvstream)
 void rstream_init(RStream *stream)
   FUNC_ATTR_NONNULL_ARG(1)
 {
-  stream->fpos = 0;
   stream->read_cb = NULL;
   stream->num_bytes = 0;
   stream->buffer = alloc_block();
@@ -156,7 +155,7 @@ static void fread_idle_cb(uv_idle_t *handle)
   stream->uvbuf.len = UV_BUF_LEN(rstream_space(stream));
 
   // Synchronous read
-  uv_fs_read(handle->loop, &req, stream->s.fd, &stream->uvbuf, 1, stream->fpos, NULL);
+  uv_fs_read(handle->loop, &req, stream->s.fd, &stream->uvbuf, 1, stream->s.fpos, NULL);
 
   uv_fs_req_cleanup(&req);
 
@@ -168,7 +167,7 @@ static void fread_idle_cb(uv_idle_t *handle)
 
   // no errors (req.result (ssize_t) is positive), it's safe to use.
   stream->write_pos += req.result;
-  stream->fpos += req.result;
+  stream->s.fpos += req.result;
   invoke_read_cb(stream, false);
 }
 

src/nvim/event/stream.c

@@ -44,6 +44,8 @@ int stream_set_blocking(int fd, bool blocking)
 void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream)
   FUNC_ATTR_NONNULL_ARG(2)
 {
+  // The underlying stream is either a file or an existing uv stream.
+  assert(uvstream == NULL ? fd >= 0 : fd < 0);
   stream->uvstream = uvstream;
 
   if (fd >= 0) {
@@ -83,6 +85,7 @@ void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream)
     stream->uvstream->data = stream;
   }
 
+  stream->fpos = 0;
   stream->internal_data = NULL;
   stream->curmem = 0;
   stream->maxmem = 0;

src/nvim/event/wstream.c

@@ -73,6 +73,26 @@ bool wstream_write(Stream *stream, WBuffer *buffer)
   // This should not be called after a stream was freed
   assert(!stream->closed);
 
+  uv_buf_t uvbuf;
+  uvbuf.base = buffer->data;
+  uvbuf.len = UV_BUF_LEN(buffer->size);
+
+  if (!stream->uvstream) {
+    uv_fs_t req;
+
+    // Synchronous write
+    uv_fs_write(stream->uv.idle.loop, &req, stream->fd, &uvbuf, 1, stream->fpos, NULL);
+
+    uv_fs_req_cleanup(&req);
+
+    wstream_release_wbuffer(buffer);
+
+    assert(stream->write_cb == NULL);
+
+    stream->fpos += MAX(req.result, 0);
+    return req.result > 0;
+  }
+
   if (stream->curmem > stream->maxmem) {
     goto err;
   }
@@ -84,10 +104,6 @@ bool wstream_write(Stream *stream, WBuffer *buffer)
   data->buffer = buffer;
   data->uv_req.data = data;
 
-  uv_buf_t uvbuf;
-  uvbuf.base = buffer->data;
-  uvbuf.len = UV_BUF_LEN(buffer->size);
-
   if (uv_write(&data->uv_req, stream->uvstream, &uvbuf, 1, write_cb)) {
     xfree(data);
     goto err;

test/functional/core/channels_spec.lua

@@ -288,6 +288,37 @@ describe('channels', function()
     eq({ 'notification', 'exit', { 3, 0 } }, next_msg())
   end)
 
+  it('stdio channel works with stdout redirected to file #30509', function()
+    t.write_file(
+      'Xstdio_write.vim',
+      [[
+        let chan = stdioopen({})
+        call chansend(chan, 'foo')
+        call chansend(chan, 'bar')
+        qall!
+      ]]
+    )
+    local fd = assert(vim.uv.fs_open('Xstdio_redir', 'w', 420))
+    local exit_code, exit_signal
+    local handle = vim.uv.spawn(nvim_prog, {
+      args = { '-u', 'NONE', '-i', 'NONE', '--headless', '-S', 'Xstdio_write.vim' },
+      -- Simulate shell redirection: "nvim ... > Xstdio_redir". #30509
+      stdio = { nil, fd, nil },
+    }, function(code, signal)
+      vim.uv.stop()
+      exit_code, exit_signal = code, signal
+    end)
+    finally(function()
+      handle:close()
+      vim.uv.fs_close(fd)
+      os.remove('Xstdio_write.vim')
+      os.remove('Xstdio_redir')
+    end)
+    vim.uv.run('default')
+    eq({ 0, 0 }, { exit_code, exit_signal })
+    eq('foobar', t.read_file('Xstdio_redir'))
+  end)
+
   it('can use buffered output mode', function()
     skip(fn.executable('grep') == 0, 'missing "grep" command')
     source([[