Fix flushed size update error on repeated flush

file/writable_file_writer.cc

@@ -860,8 +860,7 @@ IOStatus WritableFileWriter::WriteDirect(const IOOptions& opts) {
     left -= size;
     src += size;
     write_offset += size;
-    uint64_t cur_size = flushed_size_.load(std::memory_order_acquire);
-    flushed_size_.store(cur_size + size, std::memory_order_release);
+    flushed_size_.store(write_offset, std::memory_order_release);
     assert((next_write_offset_ % alignment) == 0);
   }
 
@@ -966,8 +965,7 @@ IOStatus WritableFileWriter::WriteDirectWithChecksum(const IOOptions& opts) {
 
   IOSTATS_ADD(bytes_written, left);
   assert((next_write_offset_ % alignment) == 0);
-  uint64_t cur_size = flushed_size_.load(std::memory_order_acquire);
-  flushed_size_.store(cur_size + left, std::memory_order_release);
+  flushed_size_.store(next_write_offset_ + left, std::memory_order_release);
 
   if (s.ok()) {
     // Move the tail to the beginning of the buffer

util/file_reader_writer_test.cc

@@ -334,6 +334,30 @@ TEST_F(WritableFileWriterTest, BufferWithZeroCapacityDirectIO) {
   }
 }
 
+TEST_F(WritableFileWriterTest, RepeatedFlushWithDirectIO) {
+  EnvOptions env_opts;
+  env_opts.use_direct_writes = true;
+  env_opts.writable_file_max_buffer_size = 1024 * 1024;
+
+  std::unique_ptr<WritableFileWriter> writer;
+  auto s =
+      WritableFileWriter::Create(FileSystem::Default(), /*fname=*/"dont_care_2",
+                                 FileOptions(env_opts), &writer,
+                                 /*dbg=*/nullptr);
+  EXPECT_OK(s);
+
+  const auto slice = Slice("hello");
+  EXPECT_OK(writer->Append(IOOptions(), slice));
+  ASSERT_EQ(writer->GetFileSize(), slice.size());
+
+  EXPECT_OK(writer->Flush(IOOptions()));
+  const auto flushed_size = writer->GetFlushedSize();
+
+  // flushed twice.
+  EXPECT_OK(writer->Flush(IOOptions()));
+  ASSERT_EQ(writer->GetFlushedSize(), flushed_size);
+}
+
 class DBWritableFileWriterTest : public DBTestBase {
  public:
   DBWritableFileWriterTest()