utils/file_io write_fully: assertion triggered with full disk

[andijcr]

Version & Environment

Redpanda version: (use rpk version): all

redpanda/src/v/utils/file_io.cc

Lines 54 to 73 in 6c630bd

	ss::future<> write_fully(const std::filesystem::path& p, iobuf buf) {
	static constexpr const size_t buf_size = 4096;
	auto flags = ss::open_flags::wo | ss::open_flags::create
	| ss::open_flags::truncate;
	/// Closes file on failure, otherwise file is expected to be closed in the
	/// success case where the ss::output_stream calls close()
	return ss::with_file_close_on_failure(
	ss::open_file_dma(p.string(), flags),
	[buf = std::move(buf)](ss::file f) mutable {
	return ss::make_file_output_stream(std::move(f), buf_size)
	.then([buf = std::move(buf)](ss::output_stream<char> out) mutable {
	return ss::do_with(
	std::move(out),
	[buf = std::move(buf)](ss::output_stream<char>& out) mutable {
	return write_iobuf_to_output_stream(std::move(buf), out)
	.then([&out]() mutable { return out.close(); });
	});
	});
	});
	}

write_fully(path, iobuf) can trigger an exception in case of low level failure, like no space left on disk

internally, write_fully is a uses seastar::with_file_close_on_failure(file_fut, future) and that will close file.close() if the future fails.

in our case, the internal future is ss::output_stream::close(), and the internal implementation of that is
a file.close() and rethrow of any stored exception. so

effectively that's equivalent to this code in case of internal exceptions

SEASTAR_THREAD_TEST_CASE(test_with_file_close_on_failure) {
    auto flags = ss::open_flags::rw | ss::open_flags::create
                 | ss::open_flags::truncate;
    ss::with_file_close_on_failure(
      ss::open_file_dma("/tmp/tmp.YuupbuphlR", flags),
      [](ss::file f) mutable {
          return f.close().then([] { throw "any value"; });
       })
      .get();

and this generates this trace

utils_single_thread_rpunit: seastar-prefix/src/seastar/include/seastar/core/future.hh:1917: future<T> seastar::promise<>::get_future() [T = void]: Assertion `!this->_future && this->_state && !this->_task' failed.
Aborting on shard 0.
Backtrace:
...

the solution is to move the internal operation outside. in coroutine form it's even cleaner

        auto out = co_await ss::with_file_close_on_failure(
          ss::open_file_dma("/tmp/tmp.YuupbuphlR", flags), [](ss::file f) {
              return ss::make_file_output_stream(std::move(f), buf_size);
          });
         // other code
        co_await out.close();

JIRA Link: CORE-2814