application: syncfs on storage during startup

src/v/redpanda/application.cc

@@ -351,6 +351,7 @@ int application::run(int ac, char** av) {
                 });
                 // must initialize configuration before services
                 hydrate_config(cfg);
+                sync_disks().get();
                 initialize();
                 check_environment();
                 check_for_crash_loop();
@@ -375,6 +376,27 @@ int application::run(int ac, char** av) {
     });
 }
 
+ss::future<> application::sync_disks() {
+    /*
+     * The last running instance of redpanda might have left un-flushed
+     * data or metadata in the filesystem.
+     *
+     * This does not help us in the case of e.g. IO-failed fsync() calls, which
+     * generally mark pages as clean anyway[1], but it does provide a more
+     * consistent basis for reasoning about persistence of metadata changes
+     * that might not have been fsync'd by a previous run of Redpanda.
+     *
+     * 1. https://dl.acm.org/doi/pdf/10.1145/3450338
+     */
+    vlog(_log.info, "Running syncfs on data directory...");
+    auto data_dir = co_await ss::open_directory(config::node().data_directory().as_sstring());
+    co_await data_dir.syncfs();
+
+    vlog(_log.info, "Running syncfs on cache directory...");
+    auto cache_dir = co_await ss::open_directory(config::node().cloud_storage_cache_path().string());
+    co_await cache_dir.syncfs();
+}
+
 void application::initialize(
   std::optional<YAML::Node> proxy_cfg,
   std::optional<YAML::Node> proxy_client_cfg,

src/v/redpanda/application.h

@@ -70,6 +70,7 @@ class application {
 public:
     int run(int, char**);
 
+    ss::future<> sync_disks();
     void initialize(
       std::optional<YAML::Node> proxy_cfg = std::nullopt,
       std::optional<YAML::Node> proxy_client_cfg = std::nullopt,

src/v/utils/file_sanitizer.h

@@ -134,6 +134,13 @@ class file_io_sanitizer final : public ss::file_impl {
           get_file_impl(_file)->flush());
     }
 
+    ss::future<> syncfs() final {
+        assert_file_not_closed();
+        return with_op(
+          ssx::sformat("ss::future<> syncfs(void)"),
+          get_file_impl(_file)->syncfs());
+    }
+
     ss::future<struct stat> stat() final {
         assert_file_not_closed();
         return with_op(