fs_ut.cpp

#include "fs.h"

#include "directory_handle_storage.h"
#include "log.h"
#include "loop.h"
#include "node_cache.h"

#include <cloud/filestore/libs/client/config.h>
#include <cloud/filestore/libs/client/session.h>
#include <cloud/filestore/libs/diagnostics/config.h>
#include <cloud/filestore/libs/diagnostics/critical_events.h>
#include <cloud/filestore/libs/diagnostics/filesystem_counters.h>
#include <cloud/filestore/libs/diagnostics/module_stats.h>
#include <cloud/filestore/libs/diagnostics/profile_log.h>
#include <cloud/filestore/libs/diagnostics/request_stats.h>
#include <cloud/filestore/libs/service/context.h>
#include <cloud/filestore/libs/service/filestore.h>
#include <cloud/filestore/libs/service/filestore_test.h>
#include <cloud/filestore/libs/service/request.h>
#include <cloud/filestore/libs/vfs/config.h>
#include <cloud/filestore/libs/vfs/loop.h>
#include <cloud/filestore/libs/vfs/protos/session.pb.h>
#include <cloud/filestore/libs/vfs_fuse/protos/queue_entry.pb.h>
#include <cloud/filestore/libs/vhost/client.h>
#include <cloud/filestore/libs/vhost/request.h>
#include <cloud/filestore/libs/vhost/server.h>

#include <cloud/storage/core/libs/common/error.h>
#include <cloud/storage/core/libs/common/scheduler.h>
#include <cloud/storage/core/libs/common/scheduler_test.h>
#include <cloud/storage/core/libs/common/timer.h>
#include <cloud/storage/core/libs/common/timer_test.h>
#include <cloud/storage/core/libs/diagnostics/logging.h>
#include <cloud/storage/core/libs/file_backed_containers/file_ring_buffer.h>

#include <cloud/contrib/virtiofsd/fuse.h>

#include <library/cpp/monlib/dynamic_counters/counters.h>
#include <library/cpp/testing/unittest/registar.h>

#include <util/datetime/base.h>
#include <util/folder/dirut.h>
#include <util/folder/path.h>
#include <util/folder/tempdir.h>
#include <util/generic/guid.h>
#include <util/generic/string.h>
#include <util/random/random.h>
#include <util/system/file.h>

#include <array>
#include <atomic>
#include <fstream>
#include <utility>

namespace NCloud::NFileStore::NFuse {

using namespace NThreading;

using namespace NCloud::NFileStore::NClient;
using namespace NCloud::NFileStore::NVFS;
using namespace NCloud::NFileStore::NVhost;

using namespace NCloud::NStorage::NUserStats;

namespace {

////////////////////////////////////////////////////////////////////////////////

constexpr TDuration WaitTimeout = TDuration::Seconds(5);
constexpr TDuration LargeDirectoryWaitTimeout = TDuration::Seconds(15);
constexpr TDuration ExceptionWaitTimeout = TDuration::Seconds(1);
constexpr ui64 WriteBackCacheCapacity = 1024 * 1024 + 1024;
constexpr TStringBuf MetricsComponent = "fs_ut";

// sizeof(ui64) comes from the name "." aligned up to ui64
constexpr ui64 dotSize = sizeof(fuse_direntplus) + sizeof(ui64);
constexpr ui64 dotDotSize = dotSize;

// Small directory listings for the multi-chunk paging tests below, kept tiny
// so sanitizer builds don't time out building multi-megabyte responses. Each
// entry serializes to +- 200 bytes; SecondChunkReadOffset lands past the first
// chunk (~105 KB) to page in the second.
constexpr ui32 FirstChunkEntryCount = 600;
constexpr ui32 SecondChunkEntryCount = 300;
constexpr i64 SecondChunkReadOffset = 120'000;

static const TString FileSystemId = "fs1";
static const TString SessionId = CreateGuidAsString();

static const TTempDir TempDir;

TString CreateBuffer(size_t len, char fill = 0)
{
    return TString(len, fill);
}

TString GenerateValidateData(ui32 size, ui32 seed = 0)
{
    TString data(size, 0);
    for (ui32 i = 0; i < size; ++i) {
        data[i] = 'A' + ((i + seed) % ('Z' - 'A' + 1));
    }
    return data;
}

template <class F>
bool WaitForCondition(TDuration timeout, F&& predicate)
{
    TSpinWait sw;
    auto deadline = TInstant::Now() + timeout;
    while (!predicate()) {
        if (TInstant::Now() > deadline) {
            return false;
        }
        sw.Sleep();
    }
    return true;
}

struct TTestFileMapMemoryLimiter final
    : public IFileMapMemoryLimiter
{
    std::atomic<bool> CanIncreaseResult = true;
    std::atomic<ui64> Current = 0;

    bool CanIncrease(ui64 value) const override
    {
        if (!value) {
            return true;
        }

        if (!CanIncreaseResult.load(std::memory_order_acquire)) {
            return false;
        }

        return true;
    }

    void Increase(ui64 value) override
    {
        if (value) {
            Current.fetch_add(value, std::memory_order_release);
        }
    }

    void Decrease(ui64 value) override
    {
        if (!value) {
            return;
        }

        const ui64 previous =
            Current.fetch_sub(value, std::memory_order_release);
        UNIT_ASSERT(previous >= value);
    }
};

////////////////////////////////////////////////////////////////////////////////

struct TBootstrap
{
    std::shared_ptr<TFuseVirtioClient> Fuse;

    const ILoggingServicePtr Logging;
    const ISchedulerPtr Scheduler;
    const ITimerPtr Timer;
    const NMonitoring::TDynamicCountersPtr Counters;
    const IFsCountersProviderPtr FsCountersProvider;
    const IRequestStatsRegistryPtr StatsRegistry;
    const IModuleStatsRegistryPtr ModuleStatsRegistry;

    ISessionPtr Session;
    std::shared_ptr<TFileStoreTest> Service;

    IFileSystemLoopPtr Loop;

    TString SocketPath;

    TPromise<void> StopTriggered = NewPromise<void>();

    TString DirectoryHandleStoragePath;

    TBootstrap(
            ITimerPtr timer = CreateWallClockTimer(),
            ISchedulerPtr scheduler = CreateScheduler(),
            const NProto::TFileStoreFeatures& featuresConfig = {},
            ui32 handleOpsQueueSize = 1000,
            ui32 writeBackCacheAutomaticFlushPeriodMs = 1000,
            ui64 writeBackCacheCapacity = WriteBackCacheCapacity,
            ui64 directoryHandlesInitialDataSize = 0,
            ui64 directoryHandlesMaxDataAreaStepSize = 0,
            IFileMapMemoryLimiterPtr fileMapMemoryLimiter =
                CreateFileMapMemoryLimiterStub())
        : Logging(CreateLoggingService("console", { TLOG_RESOURCES }))
        , Scheduler{std::move(scheduler)}
        , Timer{std::move(timer)}
        , Counters{MakeIntrusive<NMonitoring::TDynamicCounters>()}
        , FsCountersProvider{CreateFsCountersProvider(
            TString{MetricsComponent},
            Counters)}
        , StatsRegistry{CreateRequestStatsRegistry(
            TString{MetricsComponent},
            std::make_shared<TDiagnosticsConfig>(),
            Counters,
            Timer,
            CreateUserCounterSupplierStub(),
            FsCountersProvider)}
        , ModuleStatsRegistry{CreateModuleStatsRegistry(
            Timer,
            FsCountersProvider,
            Counters->GetSubgroup("component", TString{MetricsComponent}))}
    {
        signal(SIGUSR1, SIG_IGN);   // see fuse/driver for details

        SocketPath = (TFsPath(GetSystemTempDir()) / Sprintf("vhost.socket_%lu", RandomNumber<ui64>())).GetPath();

        InitLog(Logging);
        NVhost::InitLog(Logging);
        NVhost::StartServer();

        Fuse = std::make_shared<TFuseVirtioClient>(SocketPath, WaitTimeout);

        Service = std::make_shared<TFileStoreTest>();

        Service->CreateSessionHandler =
            [featuresConfig](auto callContext, auto request)
        {
            Y_UNUSED(callContext);

            UNIT_ASSERT(request->GetRestoreClientSession());
            NProto::TCreateSessionResponse result;
            result.MutableSession()->SetSessionId(SessionId);
            result.MutableFileStore()->SetBlockSize(4096);
            result.MutableFileStore()->MutableFeatures()->CopyFrom(
                featuresConfig);
            result.MutableFileStore()->SetFileSystemId(FileSystemId);
            return MakeFuture(result);
        };

        Service->ResetSessionHandler = [] (auto, auto) {
            return MakeFuture(NProto::TResetSessionResponse{});
        };

        Service->DestroySessionHandler = [] (auto, auto) {
            return MakeFuture(NProto::TDestroySessionResponse());
        };

        Service->PingSessionHandler = [] (auto, auto) {
            return MakeFuture(NProto::TPingSessionResponse());
        };

        auto sessionConfig = std::make_shared<TSessionConfig>(NProto::TSessionConfig{});
        Session = CreateSession(
            Logging,
            Timer,
            Scheduler,
            Service,
            std::move(sessionConfig));

        NProto::TVFSConfig proto;
        proto.SetDebug(true);
        proto.SetSocketPath(SocketPath);
        proto.SetFileSystemId(FileSystemId);
        // Keep the path configured to allow restoring
        // an existing HandleOpsQueue.
        proto.SetHandleOpsQueuePath(TempDir.Path() / "HandleOpsQueue");
        proto.SetHandleOpsQueueSize(handleOpsQueueSize);

        proto.SetDirectoryHandlesStoragePath(
            TempDir.Path() / "DirectoryHandles");
        DirectoryHandleStoragePath = proto.GetDirectoryHandlesStoragePath();
        if (featuresConfig.GetDirectoryHandlesStorageEnabled()) {
            if (directoryHandlesInitialDataSize) {
                proto.SetDirectoryHandlesInitialDataSize(
                    directoryHandlesInitialDataSize);
            }
            if (directoryHandlesMaxDataAreaStepSize) {
                proto.SetDirectoryHandlesMaxDataAreaStepSize(
                    directoryHandlesMaxDataAreaStepSize);
            }
        }

        // WriteBackCache should be configured even if it is disabled
        proto.SetWriteBackCachePath(TempDir.Path() / "WriteBackCache");
        // minimum possible capacity
        proto.SetWriteBackCacheCapacity(writeBackCacheCapacity);
        proto.SetWriteBackCacheAutomaticFlushPeriod(
            writeBackCacheAutomaticFlushPeriodMs);

        auto config = std::make_shared<TVFSConfig>(std::move(proto));
        Loop = NFuse::CreateFuseLoop(
            config,
            Logging,
            StatsRegistry,
            ModuleStatsRegistry,
            FsCountersProvider,
            Scheduler,
            Timer,
            CreateProfileLogStub(),
            Session,
            std::move(fileMapMemoryLimiter));
    }

    NMonitoring::TDynamicCountersPtr GetFileSystemStatsCounters() const
    {
        return Counters
            ->FindSubgroup("component", TString{MetricsComponent} + "_fs")
            ->FindSubgroup("host", "cluster")
            ->FindSubgroup("filesystem", FileSystemId)
            ->FindSubgroup("client", "")
            ->FindSubgroup("cloud", "")
            ->FindSubgroup("folder", "");
    }

    NMonitoring::TDynamicCountersPtr GetDirectoryHandleCounters() const
    {
        return Counters
            ->FindSubgroup("component", TString{MetricsComponent} + "_fs")
            ->FindSubgroup("host", "cluster")
            ->FindSubgroup("filesystem", FileSystemId)
            ->FindSubgroup("client", "")
            ->FindSubgroup("cloud", "")
            ->FindSubgroup("folder", "")
            ->FindSubgroup("module", "DirectoryHandles");
    }

    NMonitoring::TDynamicCountersPtr GetHandleOpsQueueCounters() const
    {
        return Counters
            ->FindSubgroup("component", TString{MetricsComponent} + "_fs")
            ->FindSubgroup("host", "cluster")
            ->FindSubgroup("filesystem", FileSystemId)
            ->FindSubgroup("client", "")
            ->FindSubgroup("cloud", "")
            ->FindSubgroup("folder", "")
            ->FindSubgroup("module", "HandleOpsQueue");
    }

    NMonitoring::TDynamicCountersPtr GetAggregateHandleOpsQueueCounters() const
    {
        return Counters
            ->FindSubgroup("component", TString{MetricsComponent})
            ->FindSubgroup("module", "HandleOpsQueue");
    }

    ~TBootstrap()
    {
        Stop();
    }

    NProto::TError Start(bool sendInitRequest=true)
    {
        auto response = StartAsync();
        UNIT_ASSERT_NO_EXCEPTION(response.Wait(WaitTimeout));
        auto error = response.GetValueSync();

        if (HasError(error)) {
            return error;
        }

        if (sendInitRequest) {
            auto init = Fuse->SendRequest<TInitRequest>();
            UNIT_ASSERT_NO_EXCEPTION(init.GetValueSync());
        }
        return error;
    }

    TFuture<NProto::TError> StartAsync()
    {
        if (Scheduler) {
            Scheduler->Start();
        }

        auto future = Loop->StartAsync();
        return future.Apply(
            [=](const auto& f)
            {
                const auto& error = f.GetValue();
                if (HasError(error)) {
                    return MakeFuture<NProto::TError>(error);
                }

                Fuse->Init();
                return MakeFuture<NProto::TError>();
            });
    }

    void Stop()
    {
        auto stop = StopAsync();
        StopTriggered.TrySetValue();

        // Scheduled tasks are not run automatically when TTestScheduler is used
        auto testScheduler = dynamic_cast<TTestScheduler*>(Scheduler.get());
        if (testScheduler) {
            auto now = TInstant::Now();
            while (!stop.HasValue() && !stop.HasException()) {
                UNIT_ASSERT(TInstant::Now() - now < WaitTimeout);
                testScheduler->RunAllScheduledTasks();
            }
        } else {
            UNIT_ASSERT(stop.Wait(WaitTimeout));
        }

        if (Scheduler) {
            // It is not allowed to stop Scheduler in the subscriber of
            // Loop->StopAsync() because the callback may be called from
            // a Scheduler thread
            Scheduler->Stop();
        }

        Fuse->DeInit();
        Loop = nullptr;
        std::remove(SocketPath.c_str());
    }

    TFuture<void> StopAsync()
    {
        return Loop ? Loop->StopAsync() : MakeFuture();
    }

    void InterruptNextRequest()
    {
        auto interrupt = Fuse->SendRequest<TInterruptRequest>(Fuse->GetLastRequestId() + 2);
        UNIT_ASSERT_NO_EXCEPTION(interrupt.GetValueSync());
    };

    static TBootstrap CreateWithHandleStorage(
        ui64 directoryHandlesInitialDataSize = 0,
        ui64 directoryHandlesMaxDataAreaStepSize = 0,
        IFileMapMemoryLimiterPtr fileMapMemoryLimiter =
            CreateFileMapMemoryLimiterStub())
    {
        NProto::TFileStoreFeatures features;
        features.SetDirectoryHandlesStorageEnabled(true);
        return TBootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features,
            1000,
            1000,
            WriteBackCacheCapacity,
            directoryHandlesInitialDataSize,
            directoryHandlesMaxDataAreaStepSize,
            std::move(fileMapMemoryLimiter));
    }
};

NProto::TNodeAttr MakeNodeAttr(ui64 nodeId, ui32 nodeType)
{
    NProto::TNodeAttr attrs;
    attrs.SetId(nodeId);
    attrs.SetType(nodeType);
    attrs.SetMode(0644);
    return attrs;
}

template <typename TSetupCreateHandler, typename TSendCreateRequest>
void DoShouldInvalidateXAttrCacheAfterSuccessfulCreate(
    TSetupCreateHandler setupCreateHandler,
    TSendCreateRequest sendCreateRequest)
{
    constexpr ui64 recycledNodeId = 6;

    NProto::TFileStoreFeatures features;
    features.SetXAttrCacheInvalidateOnCreateEnabled(true);

    auto timer = std::make_shared<TTestTimer>();
    TBootstrap bootstrap{timer, CreateScheduler(), features};
    bootstrap.Start();
    Y_DEFER {
        bootstrap.Stop();
    };

    std::atomic<int> xattrCallCount = 0;
    bootstrap.Service->SetHandlerGetNodeXAttr(
        [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            const auto call = ++xattrCallCount;
            NProto::TGetNodeXAttrResponse response;
            response.SetValue(call == 1 ? "old-value" : "new-value");
            response.SetVersion(call);
            return MakeFuture(response);
        });

    setupCreateHandler(bootstrap, recycledNodeId);

    {
        auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>(
            "name",
            recycledNodeId);
        UNIT_ASSERT(xattr.Wait(WaitTimeout));
        UNIT_ASSERT_STRINGS_EQUAL("old-value", xattr.GetValue());
    }

    sendCreateRequest(bootstrap, recycledNodeId);

    {
        auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>(
            "name",
            recycledNodeId);
        UNIT_ASSERT(xattr.Wait(WaitTimeout));
        UNIT_ASSERT_STRINGS_EQUAL("new-value", xattr.GetValue());
        UNIT_ASSERT_VALUES_EQUAL(2, xattrCallCount.load());
    }
}

}   // namespace

////////////////////////////////////////////////////////////////////////////////

Y_UNIT_TEST_SUITE(TFileSystemTest)
{
    Y_UNIT_TEST(ShouldHandleInitRequest)
    {
        TBootstrap bootstrap;
        bootstrap.Start(false);

        auto init = bootstrap.Fuse->SendRequest<TInitRequest>();
        UNIT_ASSERT_NO_EXCEPTION(init.GetValue(WaitTimeout));
    }

    void CheckWriteRequestWithFSyncQueue(bool isFSyncQueueDisabled)
    {
        NProto::TFileStoreFeatures features;
        features.SetFSyncQueueDisabled(isFSyncQueueDisabled);

        TBootstrap bootstrap(CreateWallClockTimer(), CreateScheduler(), features);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;
        bootstrap.Service->CreateHandleHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TCreateHandleResponse result;
            result.SetHandle(handleId);
            result.MutableNodeAttr()->SetId(nodeId);
            result.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
            return MakeFuture(result);
        };

        bootstrap.Service->WriteDataHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TWriteDataResponse result;
            return MakeFuture(result);
        };

        std::atomic<int> fsyncCalledWithDataSync = 0;
        std::atomic<int> fsyncCalledWithoutDataSync = 0;

        bootstrap.Service->FsyncHandler = [&] (auto callContext, auto request) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            if (request->GetDataSync()) {
                fsyncCalledWithDataSync++;
            } else {
                fsyncCalledWithoutDataSync++;
            }

            return MakeFuture(NProto::TFsyncResponse());
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto handle = bootstrap.Fuse->SendRequest<TCreateHandleRequest>("/file1", RootNodeId);
        UNIT_ASSERT_VALUES_EQUAL(handle.GetValue(WaitTimeout), handleId);

        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(
            nodeId, handleId, 0, CreateBuffer(4096, 'a'));
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));

        auto fsync = bootstrap.Fuse->SendRequest<TFsyncRequest>(
            nodeId, handleId, false /* no data sync */);
        UNIT_ASSERT_NO_EXCEPTION(fsync.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, fsyncCalledWithoutDataSync.load());
        UNIT_ASSERT_VALUES_EQUAL(0, fsyncCalledWithDataSync.load());

        fsync = bootstrap.Fuse->SendRequest<TFsyncRequest>(
            nodeId, handleId, true /* data sync */);
        UNIT_ASSERT_NO_EXCEPTION(fsync.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, fsyncCalledWithoutDataSync.load());
        UNIT_ASSERT_VALUES_EQUAL(1, fsyncCalledWithDataSync.load());
    }

    Y_UNIT_TEST(ShouldHandleWriteRequestWithFSyncQueue)
    {
        CheckWriteRequestWithFSyncQueue(false);
    }

    Y_UNIT_TEST(ShouldHandleWriteRequestWithoutFSyncQueue)
    {
        CheckWriteRequestWithFSyncQueue(true);
    }

    void CheckCreateOpenHandleRequest(
        bool isCreate,
        bool isWriteBackCacheEnabled,
        int systemFlags = O_WRONLY,
        ui32 expectedFlags = ProtoFlag(NProto::TCreateHandleRequest::E_WRITE))
    {
        NProto::TFileStoreFeatures features;
        if (isWriteBackCacheEnabled) {
            features.SetGuestWriteBackCacheEnabled(true);
        }

        auto timer = CreateWallClockTimer();
        TBootstrap bootstrap(
            timer,
            CreateScheduler(timer),
            features);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;
        bootstrap.Service->CreateHandleHandler =
            [&](auto callContext, auto request)
        {
            if (isWriteBackCacheEnabled) {
                UNIT_ASSERT(HasFlag(
                    request->GetFlags(),
                    NProto::TCreateHandleRequest::E_READ));
            } else {
                UNIT_ASSERT(!HasFlag(
                    request->GetFlags(),
                    NProto::TCreateHandleRequest::E_READ));
            }
            for (ui32 flag = NProto::TCreateHandleRequest::EFlags_MIN;
                 flag < NProto::TCreateHandleRequest::EFlags_ARRAYSIZE;
                 ++flag)
            {
                if (HasFlag(
                        expectedFlags,
                        static_cast<NProto::TCreateHandleRequest::EFlags>(
                            flag)))
                {
                    UNIT_ASSERT(HasFlag(
                        request->GetFlags(),
                        static_cast<NProto::TCreateHandleRequest::EFlags>(
                            flag)));
                }
            }
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TCreateHandleResponse result;
            result.SetHandle(handleId);
            result.MutableNodeAttr()->SetId(nodeId);
            result.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto req = std::make_shared<TCreateHandleRequest>("/file1", RootNodeId);
        req->In->Body.flags |= systemFlags;
        auto handle = bootstrap.Fuse->SendRequest<TCreateHandleRequest>(req);
        UNIT_ASSERT_VALUES_EQUAL(handle.GetValue(WaitTimeout), handleId);

        if (!isCreate) {
            auto req = std::make_shared<TOpenHandleRequest>(handleId);
            req->In->Body.flags |= systemFlags;
            auto handle = bootstrap.Fuse->SendRequest<TOpenHandleRequest>(req);
            UNIT_ASSERT_VALUES_EQUAL(handle.GetValue(WaitTimeout), handleId);
        }
    }

    Y_UNIT_TEST(ShouldHandleCreateHandleRequestWithGuestWriteBackCacheEnabled)
    {
        CheckCreateOpenHandleRequest(
            true,   // Create Handle
            false   // no guest writeback cache
        );
        CheckCreateOpenHandleRequest(
            true,   // Create Handle
            true    // guest writeback cache
        );
    }

    Y_UNIT_TEST(ShouldHandleOpenHandleRequestWithGuestWriteBackCacheEnabled)
    {
        CheckCreateOpenHandleRequest(
            false,   // Open Handle
            false    // no guest writeback cache
        );
        CheckCreateOpenHandleRequest(
            false,   // Open Handle
            true     // guest writeback cache
        );
    }

    Y_UNIT_TEST(ShouldHandleSyncFlagsInCreateAndOpenRequests)
    {
        const ui32 writeSyncFlags =
            ProtoFlag(NProto::TCreateHandleRequest::E_WRITE) |
            ProtoFlag(NProto::TCreateHandleRequest::E_SYNC);
        const ui32 writeDataSyncFlags =
            ProtoFlag(NProto::TCreateHandleRequest::E_WRITE) |
            ProtoFlag(NProto::TCreateHandleRequest::E_DSYNC);

        CheckCreateOpenHandleRequest(
            true,   // Create Handle
            false,  // no guest writeback cache
            O_WRONLY | O_SYNC,
            writeSyncFlags);
        CheckCreateOpenHandleRequest(
            false,  // Open Handle
            false,  // no guest writeback cache
            O_WRONLY | O_DSYNC,
            writeDataSyncFlags);
    }

    Y_UNIT_TEST(ShouldPassSessionId)
    {
        TBootstrap bootstrap;

        const TString sessionId = CreateGuidAsString();
        bootstrap.Service->CreateSessionHandler = [&] (auto, auto) {
            NProto::TCreateSessionResponse result;
            result.MutableSession()->SetSessionId(sessionId);
            return MakeFuture(result);
        };

        const ui64 nodeId = 123;
        const ui64 handleId = 456;
        bootstrap.Service->CreateHandleHandler = [&] (auto callContext, auto request) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TCreateHandleResponse result;
            if (GetSessionId(*request) != sessionId) {
                result = TErrorResponse(E_ARGUMENT, "invalid session id passed");
            } else {
                result.SetHandle(handleId);
                result.MutableNodeAttr()->SetId(nodeId);
                result.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
            }
            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto handle = bootstrap.Fuse->SendRequest<TCreateHandleRequest>("/file1", RootNodeId);
        UNIT_ASSERT_VALUES_EQUAL(handle.GetValue(WaitTimeout), handleId);
    }

    Y_UNIT_TEST(ShouldRecoverSession)
    {
        TBootstrap bootstrap;

        const TSet<int> expected = {100500, 100501};
        int handle = 100500;

        const TString sessionId = CreateGuidAsString();

        std::atomic_bool recovered = false;
        bootstrap.Service->CreateSessionHandler =
            [sessionId, &recovered](auto callContext, auto request)
        {
            Y_UNUSED(callContext);

            NProto::TCreateSessionResponse result;
            if (auto session = GetSessionId(*request)) {
                if (session != sessionId) {
                    NProto::TCreateSessionResponse result =
                        TErrorResponse(E_ARGUMENT, "invalid session");
                    return MakeFuture(result);
                }

                recovered = true;
            }

            result.MutableSession()->SetSessionId(sessionId);
            return MakeFuture(result);
        };

        std::atomic<int> called = 0;
        bootstrap.Service->CreateHandleHandler =
            [&](auto callContext, auto request)
        {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            called++;
            NProto::TCreateHandleResponse result;
            if (!recovered) {
                result =
                    TErrorResponse(E_FS_INVALID_SESSION, "invalid session");
            } else if (GetSessionId(*request) != sessionId) {
                result = TErrorResponse(E_ARGUMENT, "");
            } else {
                result.SetHandle(handle++);
                result.MutableNodeAttr()->SetId(100500);
                result.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
            }

            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future1 = bootstrap.Fuse->SendRequest<TCreateHandleRequest>(
            "/file1",
            RootNodeId);

        auto future2 = bootstrap.Fuse->SendRequest<TCreateHandleRequest>(
            "/file2",
            RootNodeId);

        UNIT_ASSERT(IsIn(expected, future1.GetValue(WaitTimeout)));
        UNIT_ASSERT(IsIn(expected, future2.GetValue(WaitTimeout)));
        UNIT_ASSERT(recovered);
        UNIT_ASSERT_LT(2, called.load());
    }

    Y_UNIT_TEST(ShouldFailRequestIfFailedToRecoverSession)
    {
        TBootstrap bootstrap;

        const TString sessionId = CreateGuidAsString();
        std::atomic<bool> failSession = false;

        bootstrap.Service->SetHandlerCreateSession([&] (auto, auto) {
            NProto::TCreateSessionResponse result;
            if (failSession) {
                result = TErrorResponse(E_FS_INVALID_SESSION, "invalid session");
            } else {
                result.MutableSession()->SetSessionId(sessionId);
            }
            return MakeFuture(result);
        });

        bootstrap.Service->SetHandlerCreateHandle([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TCreateHandleResponse result;
            result = TErrorResponse(E_FS_INVALID_SESSION, "");
            return MakeFuture(result);
        });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        failSession = true;
        auto future = bootstrap.Fuse->SendRequest<TCreateHandleRequest>("/file1", RootNodeId);
        UNIT_ASSERT_EXCEPTION(future.GetValue(WaitTimeout), yexception);
    }

    Y_UNIT_TEST(ShouldNoFailRequestIfSessionIsAbleToRecover)
    {
        TBootstrap bootstrap;

        const int handle = 100500;
        const TString sessionId1 = CreateGuidAsString();
        const TString sessionId2 = CreateGuidAsString();

        const TString* sessionId = &sessionId1;
        bootstrap.Service->CreateSessionHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(callContext);

            NProto::TCreateSessionResponse result;
            if (auto session = GetSessionId(*request)) {
                if (session != *sessionId) {
                    result = TErrorResponse(E_FS_INVALID_SESSION, "invalid session");
                }
            }

            result.MutableSession()->SetSessionId(*sessionId);
            sessionId = &sessionId2;

            return MakeFuture(result);
        };

        bootstrap.Service->CreateHandleHandler = [&] (auto callContext, auto request) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TCreateHandleResponse result;
            if (GetSessionId(*request) != *sessionId) {
                result = TErrorResponse(E_FS_INVALID_SESSION, "");
            } else {
                result.SetHandle(handle);
                result.MutableNodeAttr()->SetId(100500);
                result.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
            }

            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TCreateHandleRequest>("/file1", RootNodeId);
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(future.GetValue(WaitTimeout), handle);
    }

    Y_UNIT_TEST(ShouldPingSession)
    {
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler);

        const TString sessionId = CreateGuidAsString();
        bootstrap.Service->CreateSessionHandler = [&] (auto, auto) {
            NProto::TCreateSessionResponse result;
            result.MutableSession()->SetSessionId(sessionId);

            return MakeFuture(result);
        };

        bool called = false;
        bootstrap.Service->PingSessionHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(callContext);

            if (GetSessionId(*request) == sessionId) {
                called = true;
            }

            return MakeFuture(NProto::TPingSessionResponse());
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        scheduler->RunAllScheduledTasks();

        UNIT_ASSERT(called);
    }

    Y_UNIT_TEST(ShouldHandleReadDir)
    {
        TBootstrap bootstrap;
        bootstrap.Service->ListNodesHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TListNodesResponse result;
            result.AddNames()->assign("1.txt");

            auto* node = result.AddNodes();
            node->SetId(10);
            node->SetType(NProto::E_REGULAR_NODE);

            return MakeFuture(result);
        };

        std::atomic<int> fsyncDirCalledWithDataSync = 0;
        std::atomic<int> fsyncDirCalledWithoutDataSync = 0;

        bootstrap.Service->FsyncDirHandler = [&] (auto callContext, auto request) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            if (request->GetDataSync()) {
                fsyncDirCalledWithDataSync++;
            } else {
                fsyncDirCalledWithoutDataSync++;
            }

            return MakeFuture(NProto::TFsyncDirResponse());
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        const ui64 nodeId = 123;

        auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
        UNIT_ASSERT(handle.Wait(WaitTimeout));
        auto handleId = handle.GetValue();

        // read dir consists of sequantial reading until empty resposne
        auto read = bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        auto size = read.GetValue();
        UNIT_ASSERT(size > 0);

        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId, size);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        size = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size, 0);

        auto fsyncdir = bootstrap.Fuse->SendRequest<TFsyncDirRequest>(
            nodeId, handleId, false /* no data sync */);
        UNIT_ASSERT_NO_EXCEPTION(fsyncdir.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, fsyncDirCalledWithoutDataSync.load());
        UNIT_ASSERT_VALUES_EQUAL(0, fsyncDirCalledWithDataSync.load());

        fsyncdir = bootstrap.Fuse->SendRequest<TFsyncDirRequest>(
            nodeId, handleId, true /* data sync */);
        UNIT_ASSERT_NO_EXCEPTION(fsyncdir.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, fsyncDirCalledWithoutDataSync.load());
        UNIT_ASSERT_VALUES_EQUAL(1, fsyncDirCalledWithDataSync.load());

        auto close = bootstrap.Fuse->SendRequest<TReleaseDirRequest>(nodeId, handleId);
        UNIT_ASSERT_NO_EXCEPTION(close.GetValue(WaitTimeout));
    }

    Y_UNIT_TEST(ShouldHandleReadDirInvalidHandles)
    {
        TBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        const ui64 nodeId = 123;

        auto read = bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, 100500);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT(read.HasException());

        auto close = bootstrap.Fuse->SendRequest<TReleaseDirRequest>(nodeId, 100500);
        UNIT_ASSERT_NO_EXCEPTION(close.GetValue(WaitTimeout));
    }

    Y_UNIT_TEST(ShouldHandleReadDirPaging)
    {
        TBootstrap bootstrap;

        std::atomic<ui32> numCalls = 0;
        bootstrap.Service->ListNodesHandler =
            [&](auto callContext, auto request)
        {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            static ui64 id = 1;

            NProto::TListNodesResponse result;
            result.AddNames()->assign(ToString(id) + ".txt");

            auto* node = result.AddNodes();
            node->SetId(id++);
            node->SetType(NProto::E_REGULAR_NODE);

            if (!numCalls) {
                result.SetCookie("cookie");
            } else {
                Y_ABORT_UNLESS(request->GetCookie());
            }

            ++numCalls;
            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        const ui64 nodeId = 123;

        auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
        UNIT_ASSERT(handle.Wait(WaitTimeout));
        auto handleId = handle.GetValue();

        // read dir consists of sequantial reading until empty resposne
        auto read =
            bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 1);

        auto size1 = read.GetValue();
        UNIT_ASSERT(size1 > 0);

        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId,
            size1 / 2);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 1);

        auto size2 = read.GetValue();
        UNIT_ASSERT(size2 > 0);

        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId,
            size1);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 2);

        auto size3 = read.GetValue();
        UNIT_ASSERT(size3 > 0);

        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId,
            size1 + size2);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 2);

        auto size4 = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size4, 0);

        auto close =
            bootstrap.Fuse->SendRequest<TReleaseDirRequest>(nodeId, handleId);
        UNIT_ASSERT_NO_EXCEPTION(close.GetValue(WaitTimeout));
    }

    Y_UNIT_TEST(ShouldDropStoredDirectoryHandleWhenStorageExpansionIsRejected)
    {
        const TString sessionId = CreateGuidAsString();
        auto limiter = std::make_shared<TTestFileMapMemoryLimiter>();

        NProto::TFileStoreFeatures features;
        features.SetDirectoryHandlesStorageEnabled(true);

        auto createSessionHandler = [&](auto callContext, auto request)
        {
            Y_UNUSED(callContext);

            UNIT_ASSERT(request->GetRestoreClientSession());

            NProto::TCreateSessionResponse result;
            result.MutableSession()->SetSessionId(sessionId);
            result.MutableFileStore()->SetBlockSize(4096);
            result.MutableFileStore()->MutableFeatures()->CopyFrom(features);
            result.MutableFileStore()->SetFileSystemId(FileSystemId);
            return MakeFuture(result);
        };

        std::atomic<ui32> numCalls = 0;
        ui64 handleId = 0;
        TString storagePath;

        {
            auto bootstrap =
                TBootstrap::CreateWithHandleStorage(128, 128, limiter);

            bootstrap.Service->CreateSessionHandler = createSessionHandler;
            bootstrap.Service->ListNodesHandler =
                [&](auto callContext, auto request)
            {
                Y_UNUSED(request);
                UNIT_ASSERT_VALUES_EQUAL(
                    FileSystemId,
                    callContext->FileSystemId);

                ++numCalls;

                NProto::TListNodesResponse result;
                for (ui32 i = 1; i <= 1000; ++i) {
                    result.AddNames()->assign(
                        "file_with_long_name_" + ToString(i) + ".txt");
                    auto* node = result.AddNodes();
                    node->SetId(100 + i);
                    node->SetType(NProto::E_REGULAR_NODE);
                }

                return MakeFuture(result);
            };

            bootstrap.Start();
            Y_DEFER
            {
                bootstrap.Stop();
            };

            const ui64 nodeId = 123;

            auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
            UNIT_ASSERT(handle.Wait(WaitTimeout));
            handleId = handle.GetValue();

            storagePath =
                (TFsPath(bootstrap.DirectoryHandleStoragePath) / FileSystemId /
                 sessionId / "directory_handles_storage")
                    .GetPath();

            limiter->CanIncreaseResult.store(
                false,
                std::memory_order_release);

            auto read =
                bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
            UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
            const auto size = read.GetValue();
            UNIT_ASSERT_GT(size, 0);
            UNIT_ASSERT_VALUES_EQUAL(1, numCalls.load());

            read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
                nodeId,
                handleId,
                size / 2);
            UNIT_ASSERT(read.Wait(WaitTimeout));
            UNIT_ASSERT_GT(read.GetValue(), 0);
            UNIT_ASSERT_VALUES_EQUAL(1, numCalls.load());

            bootstrap.ModuleStatsRegistry->UpdateStats(true);

            auto moduleCounters = bootstrap.GetDirectoryHandleCounters();
            UNIT_ASSERT(moduleCounters);

            auto memoryLimiterRejectionCount =
                moduleCounters->FindCounter(
                    "Storage_MemoryLimiterRejectionCount");
            UNIT_ASSERT(memoryLimiterRejectionCount);
            UNIT_ASSERT_GT(memoryLimiterRejectionCount->Val(), 0);

            auto suspend = bootstrap.Loop->SuspendAsync();
            UNIT_ASSERT_NO_EXCEPTION(suspend.GetValue(WaitTimeout));
        }

        {
            auto logging = CreateLoggingService("console");
            auto log = logging->CreateLog("DIR_HANDLE_RELOAD_TEST");
            auto storage = CreateDirectoryHandleStorage(
                {.Log = log,
                 .FileMapMemoryLimiter = CreateFileMapMemoryLimiterStub(),
                 .Stats =
                     CreateDirectoryHandleStorageStats(CreateWallClockTimer()),
                 .FilePath = storagePath,
                 .MaxRecords = 100000,
                 .InitialDataAreaSize = 128,
                 .MaxDataAreaStepSize = 128,
                 .InitialDataMoveBufferSize = 1024 * 1024,
                 .PersistentHandleMaxSize = 2ull * 1024 * 1024 * 1024});

            TDirectoryHandleMap handles;
            storage->LoadHandles(handles);

            UNIT_ASSERT(!handles.contains(handleId));
        }
    }

    Y_UNIT_TEST(ShouldRemoveOrphanDirectoryHandleStorageFileWhenFeatureDisabled)
    {
        const TString sessionId = CreateGuidAsString();

        TString storageFilePath;

        // Phase 1: feature ENABLED — open a directory handle so the
        // persistent storage file is created on disk, then simulate a crash by
        // suspending the loop without the normal cleanup.
        {
            auto bootstrap = TBootstrap::CreateWithHandleStorage();

            bootstrap.Service->CreateSessionHandler =
                [&sessionId](auto callContext, auto request)
            {
                Y_UNUSED(callContext);
                UNIT_ASSERT(request->GetRestoreClientSession());

                NProto::TCreateSessionResponse result;
                result.MutableSession()->SetSessionId(sessionId);
                result.MutableFileStore()->SetBlockSize(4096);
                NProto::TFileStoreFeatures features;
                features.SetDirectoryHandlesStorageEnabled(true);
                result.MutableFileStore()->MutableFeatures()->CopyFrom(
                    features);
                result.MutableFileStore()->SetFileSystemId(FileSystemId);
                return MakeFuture(result);
            };

            bootstrap.Service->ListNodesHandler = [](auto, auto)
            {
                NProto::TListNodesResponse result;
                return MakeFuture(result);
            };

            bootstrap.Start();

            const ui64 nodeId = 123;
            auto open = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
            UNIT_ASSERT(open.Wait(WaitTimeout));
            const auto handleId = open.GetValue();

            auto read =
                bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
            UNIT_ASSERT(read.Wait(WaitTimeout));

            storageFilePath =
                (TFsPath(bootstrap.DirectoryHandleStoragePath) / FileSystemId /
                 sessionId / "directory_handles_storage")
                    .GetPath();
            UNIT_ASSERT(TFsPath(storageFilePath).Exists());

            auto suspend = bootstrap.Loop->SuspendAsync();
            UNIT_ASSERT(suspend.Wait(WaitTimeout));
        }

        // File from the previous session is still on disk.
        UNIT_ASSERT(TFsPath(storageFilePath).Exists());

        // Phase 2: feature DISABLED — startup must clean up the orphaned
        // storage file.
        {
            NProto::TFileStoreFeatures features;
            TBootstrap bootstrap(
                CreateWallClockTimer(),
                CreateScheduler(),
                features);

            bootstrap.Service->CreateSessionHandler =
                [&sessionId](auto callContext, auto request)
            {
                Y_UNUSED(callContext);
                UNIT_ASSERT(request->GetRestoreClientSession());

                NProto::TCreateSessionResponse result;
                result.MutableSession()->SetSessionId(sessionId);
                result.MutableFileStore()->SetBlockSize(4096);
                // storage disabled in the response features
                NProto::TFileStoreFeatures disabled;
                result.MutableFileStore()->MutableFeatures()->CopyFrom(
                    disabled);
                result.MutableFileStore()->SetFileSystemId(FileSystemId);
                return MakeFuture(result);
            };

            bootstrap.Start();
            Y_DEFER
            {
                bootstrap.Stop();
            };

            UNIT_ASSERT(!TFsPath(storageFilePath).Exists());
        }
    }

    Y_UNIT_TEST(ShouldHandleReadDirLargeDataWithHandlesStoragePaging)
    {
        auto bootstrap = TBootstrap::CreateWithHandleStorage();

        std::atomic<ui32> numCalls = 0;
        bootstrap.Service->ListNodesHandler =
            [&](auto callContext, auto request)
        {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            ++numCalls;

            NProto::TListNodesResponse result;

            if (!request->GetCookie()) {
                for (ui32 i = 1; i <= FirstChunkEntryCount; ++i) {
                    result.AddNames()->assign(
                        "first_chunk_file_" + ToString(i) + ".txt");
                    auto* node = result.AddNodes();
                    node->SetId(100 + i);
                    node->SetType(NProto::E_REGULAR_NODE);
                }
                result.SetCookie("has_more_data");
            } else {
                UNIT_ASSERT_VALUES_EQUAL("has_more_data", request->GetCookie());
                for (ui32 i = 1; i <= SecondChunkEntryCount; ++i) {
                    result.AddNames()->assign(
                        "second_chunk_file_" + ToString(i) + ".txt");
                    auto* node = result.AddNodes();
                    node->SetId(100 + FirstChunkEntryCount + i);
                    node->SetType(NProto::E_REGULAR_NODE);
                }
            }

            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        const ui64 nodeId = 123;

        auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
        UNIT_ASSERT(handle.Wait(WaitTimeout));
        auto handleId = handle.GetValue();

        auto read =
            bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
        UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 1);

        auto size1 = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size1, 4096);

        read =
            bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId, 0);
        UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 2);

        auto size2 = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size2, 4096);

        auto partialOffset = size1 / 2;
        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId,
            partialOffset);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 2);

        auto size3 = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size3, 4096);

        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId,
            size1);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 2);

        auto size4 = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size4, 4096);

        read =
            bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId, 0);
        UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 3);

        auto size5 = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size5, 4096);

        bootstrap.ModuleStatsRegistry->UpdateStats(true);

        auto moduleCounters = bootstrap.GetDirectoryHandleCounters();

        auto maxCacheSize = moduleCounters->FindCounter("MaxCacheSize");
        UNIT_ASSERT(maxCacheSize);
        UNIT_ASSERT_GT(maxCacheSize->Val(), 4096);

        auto maxChunkCount = moduleCounters->FindCounter("MaxChunkCount");
        UNIT_ASSERT(maxChunkCount);
        UNIT_ASSERT_VALUES_EQUAL(2, maxChunkCount->Val());

        auto maxOpenHandleCount =
            moduleCounters->FindCounter("MaxOpenHandleCount");
        UNIT_ASSERT(maxOpenHandleCount);
        UNIT_ASSERT_VALUES_EQUAL(1, maxOpenHandleCount->Val());

        // Go beyond chunk 1
        auto largeOffset = size1 + SecondChunkReadOffset;
        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId,
            largeOffset);
        UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 4);

        auto size6 = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size6, 4096);

        bootstrap.ModuleStatsRegistry->UpdateStats(true);

        maxChunkCount = moduleCounters->FindCounter("MaxChunkCount");
        UNIT_ASSERT(maxChunkCount);
        UNIT_ASSERT_VALUES_EQUAL(3, maxChunkCount->Val());

        auto firstChunkOffset = size1 / 3;
        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId,
            firstChunkOffset);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 4);

        auto size7 = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size7, 4096);

        auto secondChunkOffset = largeOffset + 200;
        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId,
            secondChunkOffset);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 4);

        auto size8 = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size8, 4096);

        // Open second directory handle for the same nodeId
        auto handle2 = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
        UNIT_ASSERT(handle2.Wait(WaitTimeout));
        auto handleId2 = handle2.GetValue();

        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId2);
        UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 5);

        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId2,
            size1);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 5);

        auto size9 = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size9, 4096);

        bootstrap.ModuleStatsRegistry->UpdateStats(true);

        UNIT_ASSERT(moduleCounters);

        maxCacheSize = moduleCounters->FindCounter("MaxCacheSize");
        UNIT_ASSERT(maxCacheSize);
        UNIT_ASSERT_GT(maxCacheSize->Val(), largeOffset);

        maxChunkCount = moduleCounters->FindCounter("MaxChunkCount");
        UNIT_ASSERT(maxChunkCount);
        UNIT_ASSERT_VALUES_EQUAL(5, maxChunkCount->Val());

        maxOpenHandleCount =
            moduleCounters->FindCounter("MaxOpenHandleCount");
        UNIT_ASSERT(maxOpenHandleCount);
        UNIT_ASSERT_VALUES_EQUAL(2, maxOpenHandleCount->Val());

        auto close2 =
            bootstrap.Fuse->SendRequest<TReleaseDirRequest>(nodeId, handleId2);
        UNIT_ASSERT_NO_EXCEPTION(close2.GetValue(WaitTimeout));

        auto close =
            bootstrap.Fuse->SendRequest<TReleaseDirRequest>(nodeId, handleId);
        UNIT_ASSERT_NO_EXCEPTION(close.GetValue(WaitTimeout));
    }

    Y_UNIT_TEST(ShouldRetainDirectoryHandlesMaxMetricsUntilWindowExpires)
    {
        auto timer = std::make_shared<TTestTimer>();

        NProto::TFileStoreFeatures features;
        features.SetDirectoryHandlesStorageEnabled(true);

        TBootstrap bootstrap{timer, CreateScheduler(), features};

        std::atomic<ui32> numCalls = 0;
        bootstrap.Service->ListNodesHandler =
            [&](auto callContext, auto request)
        {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            ++numCalls;

            NProto::TListNodesResponse result;

            if (!request->GetCookie()) {
                for (ui32 i = 1; i <= FirstChunkEntryCount; ++i) {
                    result.AddNames()->assign(
                        "first_chunk_file_" + ToString(i) + ".txt");
                    auto* node = result.AddNodes();
                    node->SetId(100 + i);
                    node->SetType(NProto::E_REGULAR_NODE);
                }
                result.SetCookie("has_more_data");
            } else {
                UNIT_ASSERT_VALUES_EQUAL("has_more_data", request->GetCookie());
                for (ui32 i = 1; i <= SecondChunkEntryCount; ++i) {
                    result.AddNames()->assign(
                        "second_chunk_file_" + ToString(i) + ".txt");
                    auto* node = result.AddNodes();
                    node->SetId(100 + FirstChunkEntryCount + i);
                    node->SetType(NProto::E_REGULAR_NODE);
                }
            }

            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER
        {
            bootstrap.Stop();
        };

        const ui64 nodeId = 123;

        auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
        UNIT_ASSERT(handle.Wait(WaitTimeout));
        auto handleId = handle.GetValue();

        auto read =
            bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
        UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 1);

        const auto size1 = read.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(size1, 4096);

        read =
            bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId, 0);
        UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 2);

        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId,
            size1 / 2);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 2);

        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId,
            size1);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 2);

        read =
            bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId, 0);
        UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 3);

        const auto largeOffset = size1 + SecondChunkReadOffset;
        read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
            nodeId,
            handleId,
            largeOffset);
        UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(numCalls.load(), 4);
        UNIT_ASSERT_VALUES_EQUAL(4096, read.GetValue());

        bootstrap.ModuleStatsRegistry->UpdateStats(true);

        auto moduleCounters = bootstrap.GetDirectoryHandleCounters();
        UNIT_ASSERT(moduleCounters);

        auto maxCacheSize = moduleCounters->FindCounter("MaxCacheSize");
        UNIT_ASSERT(maxCacheSize);
        const auto peakCacheSize = maxCacheSize->Val();
        UNIT_ASSERT_GT(peakCacheSize, largeOffset);

        auto maxChunkCount = moduleCounters->FindCounter("MaxChunkCount");
        UNIT_ASSERT(maxChunkCount);
        const auto peakChunkCount = maxChunkCount->Val();
        UNIT_ASSERT_VALUES_EQUAL(3, peakChunkCount);

        auto maxOpenHandleCount =
            moduleCounters->FindCounter("MaxOpenHandleCount");
        UNIT_ASSERT(maxOpenHandleCount);
        UNIT_ASSERT_VALUES_EQUAL(1, maxOpenHandleCount->Val());

        for (size_t i = 0; i < DirectoryHandleMaxBucketCount; ++i) {
            timer->AdvanceTime(TDuration::Seconds(1));
            bootstrap.ModuleStatsRegistry->UpdateStats(true);
        }

        maxCacheSize = moduleCounters->FindCounter("MaxCacheSize");
        UNIT_ASSERT(maxCacheSize);
        UNIT_ASSERT_VALUES_EQUAL(peakCacheSize, maxCacheSize->Val());

        maxChunkCount = moduleCounters->FindCounter("MaxChunkCount");
        UNIT_ASSERT(maxChunkCount);
        UNIT_ASSERT_VALUES_EQUAL(peakChunkCount, maxChunkCount->Val());

        auto close =
            bootstrap.Fuse->SendRequest<TReleaseDirRequest>(nodeId, handleId);
        UNIT_ASSERT_NO_EXCEPTION(close.GetValue(WaitTimeout));

        bootstrap.ModuleStatsRegistry->UpdateStats(true);

        maxCacheSize = moduleCounters->FindCounter("MaxCacheSize");
        UNIT_ASSERT(maxCacheSize);
        UNIT_ASSERT_VALUES_EQUAL(peakCacheSize, maxCacheSize->Val());

        maxChunkCount = moduleCounters->FindCounter("MaxChunkCount");
        UNIT_ASSERT(maxChunkCount);
        UNIT_ASSERT_VALUES_EQUAL(peakChunkCount, maxChunkCount->Val());

        maxOpenHandleCount =
            moduleCounters->FindCounter("MaxOpenHandleCount");
        UNIT_ASSERT(maxOpenHandleCount);
        UNIT_ASSERT_VALUES_EQUAL(1, maxOpenHandleCount->Val());

        for (size_t i = 0; i < DirectoryHandleMaxBucketCount; ++i) {
            timer->AdvanceTime(TDuration::Seconds(1));
            bootstrap.ModuleStatsRegistry->UpdateStats(true);
        }

        maxCacheSize = moduleCounters->FindCounter("MaxCacheSize");
        UNIT_ASSERT(maxCacheSize);
        UNIT_ASSERT_VALUES_EQUAL(0, maxCacheSize->Val());

        maxChunkCount = moduleCounters->FindCounter("MaxChunkCount");
        UNIT_ASSERT(maxChunkCount);
        UNIT_ASSERT_VALUES_EQUAL(0, maxChunkCount->Val());

        maxOpenHandleCount =
            moduleCounters->FindCounter("MaxOpenHandleCount");
        UNIT_ASSERT(maxOpenHandleCount);
        UNIT_ASSERT_VALUES_EQUAL(0, maxOpenHandleCount->Val());
    }

    Y_UNIT_TEST(ShouldDeallocateDirectoryHandleStorageAfterReleaseDir)
    {
        const TString sessionId = CreateGuidAsString();
        const ui64 initialDataAreaSize = 4_KB;

        NProto::TFileStoreFeatures features;
        features.SetDirectoryHandlesStorageEnabled(true);

        TBootstrap bootstrap{
            CreateWallClockTimer(),
            CreateScheduler(),
            features,
            1000,   // handleOpsQueueSize
            1000,   // writeBackCacheAutomaticFlushPeriodMs
            WriteBackCacheCapacity,
            initialDataAreaSize,
            initialDataAreaSize};

        bootstrap.Service->CreateSessionHandler = [&](auto, auto)
        {
            NProto::TCreateSessionResponse result;
            result.MutableSession()->SetSessionId(sessionId);
            result.MutableFileStore()->MutableFeatures()->CopyFrom(features);
            result.MutableFileStore()->SetFileSystemId(FileSystemId);
            return MakeFuture(result);
        };

        bootstrap.Service->ListNodesHandler = [&](auto callContext, auto)
        {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TListNodesResponse result;
            for (ui32 i = 0; i < 1024; ++i) {
                result.AddNames()->assign(
                    "dir_entry_" + ToString(i) + "_" + TString(64, 'x'));
                auto* node = result.AddNodes();
                node->SetId(1000 + i);
                node->SetType(NProto::E_REGULAR_NODE);
            }

            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER
        {
            bootstrap.Stop();
        };

        const TString pathToCache =
            (TFsPath(bootstrap.DirectoryHandleStoragePath) / FileSystemId /
             sessionId / "directory_handles_storage")
                .GetPath();

        const ui64 nodeId = 123;

        auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
        UNIT_ASSERT(handle.Wait(WaitTimeout));
        const ui64 handleId = handle.GetValue();

        const ui64 sizeBeforeRead = TFile(pathToCache, RdOnly).GetLength();

        auto read =
            bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
        UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(4096, read.GetValue());

        const ui64 sizeAfterRead = TFile(pathToCache, RdOnly).GetLength();
        UNIT_ASSERT_GT(sizeAfterRead, sizeBeforeRead);

        bootstrap.ModuleStatsRegistry->UpdateStats(true);

        auto moduleCounters = bootstrap.GetDirectoryHandleCounters();
        UNIT_ASSERT(moduleCounters);

        auto rawCapacityByteMaxCount =
            moduleCounters->FindCounter("Storage_RawCapacityByteMaxCount");
        UNIT_ASSERT(rawCapacityByteMaxCount);
        UNIT_ASSERT_GT(rawCapacityByteMaxCount->Val(), 0);

        auto storageExpansionCount =
            moduleCounters->FindCounter("Storage_ExpansionCount");
        UNIT_ASSERT(storageExpansionCount);

        auto maxOpenHandleCount =
            moduleCounters->FindCounter("MaxOpenHandleCount");
        UNIT_ASSERT(maxOpenHandleCount);
        UNIT_ASSERT_VALUES_EQUAL(1, maxOpenHandleCount->Val());

        auto rawUsedByteMaxCount =
            moduleCounters->FindCounter("Storage_RawUsedByteMaxCount");
        UNIT_ASSERT(rawUsedByteMaxCount);
        UNIT_ASSERT_GT(rawUsedByteMaxCount->Val(), 0);

        auto close =
            bootstrap.Fuse->SendRequest<TReleaseDirRequest>(nodeId, handleId);
        UNIT_ASSERT_NO_EXCEPTION(close.GetValue(WaitTimeout));

        const ui64 sizeAfterRelease = TFile(pathToCache, RdOnly).GetLength();
        UNIT_ASSERT_VALUES_EQUAL(sizeBeforeRead, sizeAfterRelease);

        bootstrap.ModuleStatsRegistry->UpdateStats(true);

        UNIT_ASSERT(moduleCounters->FindCounter("Storage_ShrinkCount"));
        UNIT_ASSERT(moduleCounters->FindCounter("Storage_CompactionCount"));
    }

    Y_UNIT_TEST(ShouldLoadDirectoryHandlesStorageWithoutErrors)
    {
        const TString sessionId = CreateGuidAsString();
        NProto::TFileStoreFeatures features;
        features.SetDirectoryHandlesStorageEnabled(true);

        auto createSessionHandler = [&](auto, auto)
        {
            NProto::TCreateSessionResponse result;
            result.MutableSession()->SetSessionId(sessionId);
            result.MutableFileStore()->MutableFeatures()->CopyFrom(features);
            result.MutableFileStore()->SetFileSystemId(FileSystemId);
            return MakeFuture(result);
        };

        std::atomic<ui32> numCalls1 = 0;
        std::atomic<ui32> numCalls2 = 0;

        TFsPath tmpPathForCache(
            TFsPath(GetSystemTempDir()) / "directory_handles_storage.dump");

        TString pathToCache;

        {
            auto bootstrap = TBootstrap::CreateWithHandleStorage();

            bootstrap.Service->CreateSessionHandler = createSessionHandler;

            bootstrap.Service->ListNodesHandler =
                [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(
                    FileSystemId,
                    callContext->FileSystemId);

                NProto::TListNodesResponse result;

                if (!request->GetCookie()) {
                    for (ui32 i = 1; i <= FirstChunkEntryCount; ++i) {
                        result.AddNames()->assign(
                            "first_chunk_file_" + ToString(i) + ".txt");
                        auto* node = result.AddNodes();
                        node->SetId(100 + i);
                        node->SetType(NProto::E_REGULAR_NODE);
                    }
                    ++numCalls1;
                    result.SetCookie("has_more_data");
                } else {
                    UNIT_ASSERT_VALUES_EQUAL(
                        "has_more_data",
                        request->GetCookie());
                    for (ui32 i = 1; i <= SecondChunkEntryCount; ++i) {
                        result.AddNames()->assign(
                            "second_chunk_file_" + ToString(i) + ".txt");
                        auto* node = result.AddNodes();
                        node->SetId(100 + FirstChunkEntryCount + i);
                        node->SetType(NProto::E_REGULAR_NODE);
                    }
                    ++numCalls2;
                }

                return MakeFuture(result);
            };

            bootstrap.Start();
            Y_DEFER
            {
                bootstrap.Stop();
            };

            pathToCache = TFsPath(bootstrap.DirectoryHandleStoragePath) /
                          FileSystemId / sessionId /
                          "directory_handles_storage";

            const ui64 nodeId = 123;

            auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
            UNIT_ASSERT(handle.Wait(WaitTimeout));
            auto handleId = handle.GetValue();

            auto read =
                bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
            UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
            UNIT_ASSERT_VALUES_EQUAL(numCalls1.load(), 1);
            UNIT_ASSERT_VALUES_EQUAL(numCalls2.load(), 0);

            // Go beyond the first chunk
            auto largeOffset = SecondChunkReadOffset;
            read = bootstrap.Fuse->SendRequest<TReadDirRequest>(
                nodeId,
                handleId,
                largeOffset);
            UNIT_ASSERT(read.Wait(LargeDirectoryWaitTimeout));
            UNIT_ASSERT_VALUES_EQUAL(numCalls1.load(), 1);
            UNIT_ASSERT_VALUES_EQUAL(numCalls2.load(), 1);

            TFsPath(pathToCache).CopyTo(tmpPathForCache.GetPath(), true);
        }

        {
            auto bootstrap = TBootstrap::CreateWithHandleStorage();

            tmpPathForCache.CopyTo(pathToCache, true);

            bootstrap.Service->CreateSessionHandler = createSessionHandler;

            bootstrap.Start();
            bootstrap.Stop();
        }
    }

    Y_UNIT_TEST(ShouldNotAllowGuestCachingForRestoredDirectoryContent)
    {
        const TString sessionId = CreateGuidAsString();
        NProto::TFileStoreFeatures features;
        features.SetDirectoryHandlesStorageEnabled(true);

        auto createSessionHandler = [&](auto, auto)
        {
            NProto::TVfsSessionState state;
            state.SetProtoMajor(7);
            state.SetProtoMinor(33);
            state.SetBufferSize(256 * 1024);

            NProto::TCreateSessionResponse result;
            result.MutableSession()->SetSessionId(sessionId);
            UNIT_ASSERT(state.SerializeToString(
                result.MutableSession()->MutableSessionState()));
            result.MutableFileStore()->MutableFeatures()->CopyFrom(features);
            result.MutableFileStore()->SetFileSystemId(FileSystemId);
            return MakeFuture(result);
        };

        std::atomic<ui32> numCalls = 0;
        auto listNodesHandler = [&](auto, auto)
        {
            ++numCalls;

            NProto::TListNodesResponse result;
            for (ui32 i = 1; i <= 10; ++i) {
                result.AddNames()->assign("file_" + ToString(i) + ".txt");
                auto* node = result.AddNodes();
                node->SetId(100 + i);
                node->SetType(NProto::E_REGULAR_NODE);
            }
            return MakeFuture(result);
        };

        const ui64 nodeId = 123;
        ui64 handleId = 0;

        TFsPath tmpPathForCache(
            TFsPath(GetSystemTempDir()) / "directory_handles_storage.restore");
        TString pathToCache;

        {
            auto bootstrap = TBootstrap::CreateWithHandleStorage();

            bootstrap.Service->CreateSessionHandler = createSessionHandler;
            bootstrap.Service->ListNodesHandler = listNodesHandler;

            bootstrap.Start();
            Y_DEFER
            {
                bootstrap.Stop();
            };

            pathToCache = TFsPath(bootstrap.DirectoryHandleStoragePath) /
                          FileSystemId / sessionId /
                          "directory_handles_storage";

            auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
            UNIT_ASSERT(handle.Wait(WaitTimeout));
            handleId = handle.GetValue();

            auto read =
                bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
            UNIT_ASSERT(read.Wait(WaitTimeout));
            UNIT_ASSERT_GT(read.GetValue(), 0);
            UNIT_ASSERT_VALUES_EQUAL(1, numCalls.load());

            TFsPath(pathToCache).CopyTo(tmpPathForCache.GetPath(), true);
        }

        {
            auto bootstrap = TBootstrap::CreateWithHandleStorage();

            tmpPathForCache.CopyTo(pathToCache, true);

            bootstrap.Service->CreateSessionHandler = createSessionHandler;
            bootstrap.Service->ListNodesHandler = listNodesHandler;

            //
            // The guest doesn't reinitialize the connection after a vhost
            // restart, so the restored directory content is served without
            // FUSE_INIT.
            //

            bootstrap.Start(false /* sendInitRequest */);
            Y_DEFER
            {
                bootstrap.Stop();
            };

            auto rq = std::make_shared<TReadDirRequest>(
                nodeId,
                handleId,
                dotSize + dotDotSize);
            auto rf = bootstrap.Fuse->SendRequest(rq);
            UNIT_ASSERT_NO_EXCEPTION(rf.GetValue(WaitTimeout));

            // Restored content should be served from the handle cache
            UNIT_ASSERT_VALUES_EQUAL(1, numCalls.load());

            auto buf = TStringBuf(
                reinterpret_cast<const char*>(rq->Out->Data()),
                rq->Out->Header.len - sizeof(rq->Out->Header));
            UNIT_ASSERT_LE(sizeof(fuse_direntplus), buf.size());
            const auto* de =
                reinterpret_cast<const fuse_direntplus*>(buf.data());

            //
            // Restored content was persisted before the restart and the cache
            // versions are lost, so neither entries nor attrs can be trusted
            // to be up-to-date.
            //

            UNIT_ASSERT_VALUES_EQUAL(101, de->entry_out.nodeid);
            UNIT_ASSERT_VALUES_EQUAL(0, de->entry_out.entry_valid);
            UNIT_ASSERT_VALUES_EQUAL(0, de->entry_out.entry_valid_nsec);
            UNIT_ASSERT_VALUES_EQUAL(0, de->entry_out.attr_valid);
            UNIT_ASSERT_VALUES_EQUAL(0, de->entry_out.attr_valid_nsec);
        }
    }

    Y_UNIT_TEST(ShouldHandleForgetRequestsForUnknownNodes)
    {
        TBootstrap bootstrap;

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        const ui64 nodeId = 123;
        const ui64 refCount = 10;

        auto forget = bootstrap.Fuse->SendRequest<TForgetRequest>(nodeId, refCount);
        UNIT_ASSERT_NO_EXCEPTION(forget.GetValue(WaitTimeout));
    }

    Y_UNIT_TEST(ShouldResetSessionStateUponInitAndDestroy)
    {
        TBootstrap bootstrap;

        TMutex stateMutex;
        TString state;
        std::atomic<ui32> resets = 0;
        bootstrap.Service->ResetSessionHandler = [&] (auto callContext, auto request) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            ++resets;
            with_lock(stateMutex) {
                state = request->GetSessionState();
            }

            return MakeFuture(NProto::TResetSessionResponse{});
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        UNIT_ASSERT_VALUES_EQUAL(resets.load(), 1);
        with_lock(stateMutex) {
            UNIT_ASSERT(state);
        }

        auto destroy = bootstrap.Fuse->SendRequest<TDestroyRequest>();
        UNIT_ASSERT_NO_EXCEPTION(destroy.GetValue(WaitTimeout));

        UNIT_ASSERT_VALUES_EQUAL(resets.load(), 2);
        with_lock(stateMutex) {
            UNIT_ASSERT(!state);
        }
    }

    Y_UNIT_TEST(ShouldReinitSessionWithoutInitRequest)
    {
        TBootstrap bootstrap;

        const ui64 nodeId = 123;
        const ui64 handleId = 456;
        bootstrap.Service->CreateHandleHandler = [&] (auto callContext, auto request) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            Y_ABORT_UNLESS(request->GetNodeId() != nodeId);

            NProto::TCreateHandleResponse response;
            response.SetHandle(handleId);
            response.MutableNodeAttr()->SetId(nodeId);
            response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);

            return MakeFuture(response);
        };

        bootstrap.Service->CreateSessionHandler = [&] (auto, auto) {
            NProto::TVfsSessionState state;
            state.SetProtoMajor(7);
            state.SetProtoMinor(33);
            state.SetBufferSize(256 * 1024);

            NProto::TCreateSessionResponse response;
            response.MutableSession()->SetSessionId(CreateGuidAsString());
            Y_ABORT_UNLESS(state.SerializeToString(response.MutableSession()->MutableSessionState()));

            return MakeFuture(response);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto handle = bootstrap.Fuse->SendRequest<TCreateHandleRequest>("/file1", RootNodeId);
        UNIT_ASSERT_VALUES_EQUAL(handle.GetValue(WaitTimeout), handleId);
    }

    Y_UNIT_TEST(ShouldWaitToAcquireLock)
    {
        TBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        std::atomic<ui32> called = 0;
        bootstrap.Service->SetHandlerAcquireLock([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            if (++called < 3) {
                NProto::TAcquireLockResponse response = TErrorResponse(E_FS_WOULDBLOCK, "xxx");
                return MakeFuture(response);
            }

            NProto::TAcquireLockResponse response;
            return MakeFuture(response);
        });

        auto result = bootstrap.Fuse->SendRequest<TAcquireLockRequest>(0, F_RDLCK);
        UNIT_ASSERT_NO_EXCEPTION(result.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(called.load(), 3);
    }

    Y_UNIT_TEST(ShouldNotWaitToAcquireLock)
    {
        TBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        std::atomic<ui32> called = 0;
        bootstrap.Service->SetHandlerAcquireLock([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            ++called;
            NProto::TAcquireLockResponse response = TErrorResponse(E_FS_WOULDBLOCK, "xxx");
            return MakeFuture(response);
        });

        auto result = bootstrap.Fuse->SendRequest<TAcquireLockRequest>(0, F_RDLCK, false);
        UNIT_ASSERT(result.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(called.load(), 1);
        UNIT_ASSERT(result.HasException());

        bootstrap.Service->SetHandlerAcquireLock([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            ++called;
            NProto::TAcquireLockResponse response = TErrorResponse(E_FS_BADHANDLE, "xxx");
            return MakeFuture(response);
        });

        result = bootstrap.Fuse->SendRequest<TAcquireLockRequest>(0, F_RDLCK);
        UNIT_ASSERT(result.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(called.load(), 2);
        UNIT_ASSERT(result.HasException());
    }

    Y_UNIT_TEST(ShouldTestLock)
    {
        TBootstrap bootstrap;
        bootstrap.Start();
        static constexpr pid_t DefaultPid = 123;

        bootstrap.Service->SetHandlerTestLock([&] (auto callContext, auto /* request */) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TTestLockResponse response;
            return MakeFuture(response);
        });

        auto result = bootstrap.Fuse->SendRequest<TTestLockRequest>(0, F_RDLCK, 0, 100);
        UNIT_ASSERT(result.Wait(WaitTimeout));
        auto lk = result.GetValue();
        UNIT_ASSERT_VALUES_EQUAL(lk.type, F_UNLCK);

        bootstrap.Service->SetHandlerTestLock([&] (auto callContext, auto /* request */) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TTestLockResponse response;
            return MakeFuture(response);
        });

        bootstrap.Service->SetHandlerTestLock([&] (auto callContext, auto /* request */) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TTestLockResponse response = TErrorResponse(E_FS_WOULDBLOCK, "");
            response.SetOwner(100);
            response.SetOffset(100500);
            response.SetLength(500100);
            response.SetLockType(NProto::E_EXCLUSIVE);
            response.SetPid(DefaultPid);

            return MakeFuture(response);
        });

        result = bootstrap.Fuse->SendRequest<TTestLockRequest>(0, F_RDLCK, 0, 100);
        UNIT_ASSERT(result.Wait(WaitTimeout));
        lk = result.GetValue();

        UNIT_ASSERT_VALUES_EQUAL(lk.type, F_WRLCK);
        UNIT_ASSERT_VALUES_EQUAL(lk.start, 100500);
        UNIT_ASSERT_VALUES_EQUAL(lk.end, 100500 + 500100 - 1);
        UNIT_ASSERT_VALUES_EQUAL(lk.pid, DefaultPid);

        bootstrap.Service->SetHandlerTestLock([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TTestLockResponse response = TErrorResponse(E_FS_BADHANDLE, "");
            return MakeFuture(response);
        });

        result = bootstrap.Fuse->SendRequest<TTestLockRequest>(0, F_RDLCK, 0, 100);
        UNIT_ASSERT(result.Wait(WaitTimeout));
        UNIT_ASSERT_EXCEPTION(result.GetValue(), yexception);
    }

    Y_UNIT_TEST(ShouldNotFailOnSuspendWithRequestsInFlight)
    {
        std::atomic<bool> sessionDestroyed = false;

        TBootstrap bootstrap;
        bootstrap.Service->DestroySessionHandler = [&sessionDestroyed] (auto, auto) {
            sessionDestroyed = true;
            return MakeFuture(NProto::TDestroySessionResponse());
        };

        auto response = NewPromise<NProto::TListNodesResponse>();
        bootstrap.Service->ListNodesHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            return response.GetFuture();
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        const ui64 nodeId = 123;

        auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
        UNIT_ASSERT(handle.Wait(WaitTimeout));
        auto handleId = handle.GetValue();

        auto read =
            bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
        UNIT_ASSERT(!read.Wait(TDuration::Seconds(1)));

        auto suspend = bootstrap.Loop->SuspendAsync();
        UNIT_ASSERT(!suspend.Wait(TDuration::Seconds(1)));

        response.SetValue(NProto::TListNodesResponse{});
        UNIT_ASSERT(suspend.Wait(WaitTimeout));
        UNIT_ASSERT_NO_EXCEPTION(read.GetValueSync());
    }

    Y_UNIT_TEST(ShouldNotFailOnStopWithRequestsInFlight)
    {
        std::atomic<bool> sessionDestroyed = false;

        TBootstrap bootstrap;
        bootstrap.Service->DestroySessionHandler = [&sessionDestroyed] (auto, auto) {
            sessionDestroyed = true;
            return MakeFuture(NProto::TDestroySessionResponse());
        };

        auto response = NewPromise<NProto::TListNodesResponse>();
        bootstrap.Service->ListNodesHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            return response.GetFuture();
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        const ui64 nodeId = 123;

        auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
        UNIT_ASSERT(handle.Wait(WaitTimeout));
        auto handleId = handle.GetValue();

        auto read =
            bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
        UNIT_ASSERT(!read.Wait(TDuration::Seconds(1)));

        auto stop = bootstrap.StopAsync();
        UNIT_ASSERT(!stop.Wait(TDuration::Seconds(1)));

        auto read2 =
            bootstrap.Fuse->SendRequest<TReadDirRequest>(nodeId, handleId);
        UNIT_ASSERT_EXCEPTION_CONTAINS(
            read2.GetValueSync(),
            yexception,
            "Unknown error -4");

        response.SetValue(NProto::TListNodesResponse{});
        UNIT_ASSERT(stop.Wait(WaitTimeout));
        UNIT_ASSERT_NO_EXCEPTION(read.GetValueSync());
    }

    Y_UNIT_TEST(ShouldNotAbortOnInvalidServerLookup)
    {
        TBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        bootstrap.Service->SetHandlerGetNodeAttr([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TGetNodeAttrResponse response;
            // response.MutableNode()->SetId(123);
            return MakeFuture(response);
        });

        auto lookup = bootstrap.Fuse->SendRequest<TLookupRequest>("test", RootNodeId);
        UNIT_ASSERT_EXCEPTION_CONTAINS(
            lookup.GetValue(WaitTimeout),
            yexception,
            "Unknown error -5");
    }

    Y_UNIT_TEST(ShouldCacheXAttrValueOnGet)
    {
        TBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        std::atomic<int> callCount = 0;
        bootstrap.Service->SetHandlerGetNodeXAttr([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            ++callCount;
            NProto::TGetNodeXAttrResponse response;
            response.SetValue("value");
            response.SetVersion(1);
            return MakeFuture(response);
        });

        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>("name", 6);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_STRINGS_EQUAL("value", xattr.GetValue());
        }
        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>("name", 6);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_STRINGS_EQUAL("value", xattr.GetValue());
            UNIT_ASSERT_VALUES_EQUAL(1, callCount.load());
        }
    }

    Y_UNIT_TEST(ShouldCacheXAttrValueOnSet)
    {
        TBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        int callCount = 0;
        bootstrap.Service->SetHandlerGetNodeXAttr([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            ++callCount;
            NProto::TGetNodeXAttrResponse response;
            return MakeFuture(response);
        });
        bootstrap.Service->SetHandlerSetNodeXAttr([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TSetNodeXAttrResponse response;
            response.SetVersion(1);
            return MakeFuture(response);
        });

        {
            auto set = bootstrap.Fuse->SendRequest<TSetXAttrValueRequest>("name", "value", RootNodeId);
            UNIT_ASSERT(set.Wait(WaitTimeout));
        }
        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>("name", RootNodeId);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_STRINGS_EQUAL("value", xattr.GetValue());
            UNIT_ASSERT_VALUES_EQUAL(0, callCount);
        }
    }

    Y_UNIT_TEST(ShouldSkipCacheValueAfterTimeout)
    {
        std::shared_ptr<TTestTimer> timer = std::make_shared<TTestTimer>();
        TBootstrap bootstrap{timer};
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        std::atomic<int> callCount = 0;
        bootstrap.Service->SetHandlerGetNodeXAttr([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            ++callCount;
            NProto::TGetNodeXAttrResponse response;
            response.SetValue("value");
            response.SetVersion(callCount);
            return MakeFuture(response);
        });

        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>("name", 6);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_STRINGS_EQUAL("value", xattr.GetValue());
        }

        timer->AdvanceTime(TDuration::Hours(1));

        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>("name", 6);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_STRINGS_EQUAL("value", xattr.GetValue());
            UNIT_ASSERT_VALUES_EQUAL(2, callCount.load());
        }
    }

    Y_UNIT_TEST(ShouldNotCacheXAttrWhenErrorHappens)
    {
        TBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        std::atomic<int> callCount = 0;
        bootstrap.Service->SetHandlerGetNodeXAttr([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TGetNodeXAttrResponse response;
            if (callCount.load() == 0) {
                response.MutableError()->SetCode(MAKE_FILESTORE_ERROR(NProto::E_FS_XATTR2BIG));
            } else {
                response.SetValue("value");
                response.SetVersion(callCount.load());
            }
            ++callCount;
            return MakeFuture(response);
        });

        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>("name", 6);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_EXCEPTION(xattr.GetValue(), yexception);
        }
        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>("name", 6);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_STRINGS_EQUAL("value", xattr.GetValue());
            UNIT_ASSERT_VALUES_EQUAL(2, callCount.load());
        }
    }

    Y_UNIT_TEST(ShouldCacheXAttrAbsence)
    {
        TBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        std::atomic<int> callCount = 0;
        bootstrap.Service->SetHandlerGetNodeXAttr([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TGetNodeXAttrResponse response;
            response.MutableError()->SetCode(MAKE_FILESTORE_ERROR(NProto::E_FS_NOXATTR));
            ++callCount;
            return MakeFuture(response);
        });

        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>("name", 6);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_EXCEPTION_CONTAINS(
                xattr.GetValue(),
                yexception,
                "-61"); // NODATA error code
        }
        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>("name", 6);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_EXCEPTION_CONTAINS(
                xattr.GetValue(),
                yexception,
                "-61"); // NODATA error code
            UNIT_ASSERT_VALUES_EQUAL(1, callCount.load());
        }
    }

    Y_UNIT_TEST(ShouldInvalidateXAttrCacheAfterMkDirReusesNodeId)
    {
        DoShouldInvalidateXAttrCacheAfterSuccessfulCreate(
            [] (TBootstrap& bootstrap, ui64 recycledNodeId) {
                bootstrap.Service->SetHandlerCreateNode(
                    [=] (auto callContext, auto request) {
                        Y_UNUSED(request);
                        UNIT_ASSERT_VALUES_EQUAL(
                            FileSystemId,
                            callContext->FileSystemId);

                        NProto::TCreateNodeResponse response;
                        *response.MutableNode() = MakeNodeAttr(
                            recycledNodeId,
                            NProto::E_DIRECTORY_NODE);
                        return MakeFuture(response);
                    });
            },
            [] (TBootstrap& bootstrap, ui64 recycledNodeId) {
                Y_UNUSED(recycledNodeId);
                auto create = bootstrap.Fuse->SendRequest<TMkDirRequest>(
                    "dir",
                    RootNodeId);
                UNIT_ASSERT(create.Wait(WaitTimeout));
            });
    }

    Y_UNIT_TEST(ShouldInvalidateXAttrCacheAfterMkNodeReusesNodeId)
    {
        DoShouldInvalidateXAttrCacheAfterSuccessfulCreate(
            [] (TBootstrap& bootstrap, ui64 recycledNodeId) {
                bootstrap.Service->SetHandlerCreateNode(
                    [=] (auto callContext, auto request) {
                        Y_UNUSED(request);
                        UNIT_ASSERT_VALUES_EQUAL(
                            FileSystemId,
                            callContext->FileSystemId);

                        NProto::TCreateNodeResponse response;
                        *response.MutableNode() = MakeNodeAttr(
                            recycledNodeId,
                            NProto::E_REGULAR_NODE);
                        return MakeFuture(response);
                    });
            },
            [] (TBootstrap& bootstrap, ui64 recycledNodeId) {
                Y_UNUSED(recycledNodeId);
                auto create = bootstrap.Fuse->SendRequest<TMkNodeRequest>(
                    "file",
                    RootNodeId);
                UNIT_ASSERT(create.Wait(WaitTimeout));
            });
    }

    Y_UNIT_TEST(ShouldInvalidateXAttrCacheAfterSymLinkReusesNodeId)
    {
        DoShouldInvalidateXAttrCacheAfterSuccessfulCreate(
            [] (TBootstrap& bootstrap, ui64 recycledNodeId) {
                bootstrap.Service->SetHandlerCreateNode(
                    [=] (auto callContext, auto request) {
                        Y_UNUSED(request);
                        UNIT_ASSERT_VALUES_EQUAL(
                            FileSystemId,
                            callContext->FileSystemId);

                        NProto::TCreateNodeResponse response;
                        *response.MutableNode() = MakeNodeAttr(
                            recycledNodeId,
                            NProto::E_LINK_NODE);
                        return MakeFuture(response);
                    });
            },
            [] (TBootstrap& bootstrap, ui64 recycledNodeId) {
                Y_UNUSED(recycledNodeId);
                auto create = bootstrap.Fuse->SendRequest<TSymLinkRequest>(
                    "link",
                    "target",
                    RootNodeId);
                UNIT_ASSERT(create.Wait(WaitTimeout));
            });
    }

    Y_UNIT_TEST(ShouldInvalidateXAttrCacheAfterCreateHandleReusesNodeId)
    {
        DoShouldInvalidateXAttrCacheAfterSuccessfulCreate(
            [] (TBootstrap& bootstrap, ui64 recycledNodeId) {
                bootstrap.Service->SetHandlerCreateHandle(
                    [=] (auto callContext, auto request) {
                        Y_UNUSED(request);
                        UNIT_ASSERT_VALUES_EQUAL(
                            FileSystemId,
                            callContext->FileSystemId);

                        NProto::TCreateHandleResponse response;
                        response.SetHandle(42);
                        *response.MutableNodeAttr() = MakeNodeAttr(
                            recycledNodeId,
                            NProto::E_REGULAR_NODE);
                        return MakeFuture(response);
                    });
            },
            [] (TBootstrap& bootstrap, ui64 recycledNodeId) {
                Y_UNUSED(recycledNodeId);
                auto request = std::make_shared<TCreateHandleRequest>(
                    "file",
                    RootNodeId);
                request->In->Body.flags = O_CREAT | O_RDWR;
                request->In->Body.mode = 0644;

                auto create = bootstrap.Fuse->SendRequest(request);
                UNIT_ASSERT(create.Wait(WaitTimeout));
            });
    }

    Y_UNIT_TEST(ShouldNotInvalidateXAttrCacheAfterFailedCreate)
    {
        constexpr ui64 recycledNodeId = 6;

        NProto::TFileStoreFeatures features;
        features.SetXAttrCacheInvalidateOnCreateEnabled(true);

        auto timer = std::make_shared<TTestTimer>();
        TBootstrap bootstrap{timer, CreateScheduler(), features};
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        std::atomic<int> xattrCallCount = 0;
        bootstrap.Service->SetHandlerGetNodeXAttr(
            [&] (auto callContext, auto request) {
                Y_UNUSED(request);
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

                ++xattrCallCount;
                NProto::TGetNodeXAttrResponse response;
                response.SetValue("old-value");
                response.SetVersion(xattrCallCount.load());
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerCreateNode(
            [] (auto callContext, auto request) {
                Y_UNUSED(request);
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

                NProto::TCreateNodeResponse response;
                response.MutableError()->SetCode(MAKE_FILESTORE_ERROR(
                    NProto::E_FS_IO));
                return MakeFuture(response);
            });

        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>(
                "name",
                recycledNodeId);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_STRINGS_EQUAL("old-value", xattr.GetValue());
        }

        {
            auto create = bootstrap.Fuse->SendRequest<TMkNodeRequest>(
                "file",
                RootNodeId);
            UNIT_ASSERT(create.Wait(WaitTimeout));
            UNIT_ASSERT_EXCEPTION(create.GetValue(), yexception);
        }

        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>(
                "name",
                recycledNodeId);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_STRINGS_EQUAL("old-value", xattr.GetValue());
            UNIT_ASSERT_VALUES_EQUAL(1, xattrCallCount.load());
        }
    }

    Y_UNIT_TEST(ShouldNotInvalidateXAttrCacheAfterLookupLinkOrOpenExisting)
    {
        constexpr ui64 nodeId = 6;

        NProto::TFileStoreFeatures features;
        features.SetXAttrCacheInvalidateOnCreateEnabled(true);

        auto timer = std::make_shared<TTestTimer>();
        TBootstrap bootstrap{timer, CreateScheduler(), features};
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        std::atomic<int> xattrCallCount = 0;
        bootstrap.Service->SetHandlerGetNodeXAttr(
            [&] (auto callContext, auto request) {
                Y_UNUSED(request);
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

                ++xattrCallCount;
                NProto::TGetNodeXAttrResponse response;
                response.SetValue("old-value");
                response.SetVersion(xattrCallCount.load());
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerGetNodeAttr(
            [=] (auto callContext, auto request) {
                Y_UNUSED(request);
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

                NProto::TGetNodeAttrResponse response;
                *response.MutableNode() = MakeNodeAttr(
                    nodeId,
                    NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerCreateNode(
            [=] (auto callContext, auto request) {
                Y_UNUSED(request);
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

                NProto::TCreateNodeResponse response;
                *response.MutableNode() = MakeNodeAttr(
                    nodeId,
                    NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerCreateHandle(
            [=] (auto callContext, auto request) {
                Y_UNUSED(request);
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

                NProto::TCreateHandleResponse response;
                response.SetHandle(42);
                *response.MutableNodeAttr() = MakeNodeAttr(
                    nodeId,
                    NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        auto assertCachedXAttr = [&] {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>(
                "name",
                nodeId);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_STRINGS_EQUAL("old-value", xattr.GetValue());
            UNIT_ASSERT_VALUES_EQUAL(1, xattrCallCount.load());
        };

        {
            auto xattr = bootstrap.Fuse->SendRequest<TGetXAttrValueRequest>(
                "name",
                nodeId);
            UNIT_ASSERT(xattr.Wait(WaitTimeout));
            UNIT_ASSERT_STRINGS_EQUAL("old-value", xattr.GetValue());
        }

        {
            auto lookup = bootstrap.Fuse->SendRequest<TLookupRequest>(
                "file",
                RootNodeId);
            UNIT_ASSERT(lookup.Wait(WaitTimeout));
            assertCachedXAttr();
        }

        {
            auto link = bootstrap.Fuse->SendRequest<TLinkRequest>(
                "hardlink",
                nodeId,
                RootNodeId);
            UNIT_ASSERT(link.Wait(WaitTimeout));
            assertCachedXAttr();
        }

        {
            auto request = std::make_shared<TCreateHandleRequest>(
                "file",
                RootNodeId);
            request->In->Body.flags = O_RDWR;

            auto create = bootstrap.Fuse->SendRequest(request);
            UNIT_ASSERT(create.Wait(WaitTimeout));
            assertCachedXAttr();
        }
    }

    Y_UNIT_TEST(SuspendShouldNotDestroySession)
    {
        TBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        std::atomic<bool> sessionDestroyed = false;
        bootstrap.Service->SetHandlerDestroySession([&] (auto, auto) {
            sessionDestroyed = true;
            NProto::TDestroySessionResponse response;
            return MakeFuture(response);
        });

        bootstrap.Service->SetHandlerResetSession([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            sessionDestroyed = true;
            NProto::TResetSessionResponse response;
            return MakeFuture(response);
        });

        auto future = bootstrap.Loop->SuspendAsync();
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
        UNIT_ASSERT(!sessionDestroyed);

        // prevent stack use after scope on shutdown
        bootstrap.Service->SetHandlerResetSession([&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            return MakeFuture(NProto::TResetSessionResponse{});
        });

        bootstrap.Service->SetHandlerDestroySession([] (auto, auto) {
            return MakeFuture(NProto::TDestroySessionResponse());
        });
    }

    Y_UNIT_TEST(StopShouldDestroySession)
    {
        bool sessionReset = false;
        bool sessionDestroyed = false;

        TBootstrap bootstrap;
        bootstrap.Service->ResetSessionHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            sessionReset = true;
            NProto::TResetSessionResponse response;
            return MakeFuture(response);
        };
        bootstrap.Service->DestroySessionHandler = [&] (auto, auto) {
            sessionDestroyed = true;
            NProto::TDestroySessionResponse response;
            return MakeFuture(response);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Loop->StopAsync();
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
        UNIT_ASSERT(sessionReset);
        UNIT_ASSERT(sessionDestroyed);

        // prevent stack use after scope on shutdown
        bootstrap.Service->ResetSessionHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            return MakeFuture(NProto::TResetSessionResponse{});
        };

        bootstrap.Service->DestroySessionHandler = [] (auto, auto) {
            return MakeFuture(NProto::TDestroySessionResponse());
        };
    }

    Y_UNIT_TEST(ShouldHitErrorMetricOnFailure)
    {
        TBootstrap bootstrap;

        bootstrap.Service->CreateHandleHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TCreateHandleResponse result = TErrorResponse(E_FS_EXIST, "");
            return MakeFuture(result);
        };

        bootstrap.Start();

        auto future = bootstrap.Fuse->SendRequest<TCreateHandleRequest>("/file1", RootNodeId);
        UNIT_ASSERT_EXCEPTION(future.GetValueSync(), yexception);
        bootstrap.Stop(); // wait till all requests are done writing their stats

        auto counters = bootstrap.Counters
            ->FindSubgroup("component", "fs_ut")
            ->FindSubgroup("request", "CreateHandle");
        UNIT_ASSERT_VALUES_EQUAL(1, counters->GetCounter("Errors")->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, counters->GetCounter("Errors/Fatal")->GetAtomic());
    }

    Y_UNIT_TEST(ShouldNotCrashWhenStoppedBeforeFileStoreResponse)
    {
        return; // NBS-4767
        TBootstrap bootstrap;

        auto create = NewPromise<NProto::TCreateHandleResponse>();
        bootstrap.Service->CreateHandleHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            return create.GetFuture();
        };

        bootstrap.Start();

        auto future = bootstrap.Fuse->SendRequest<TCreateHandleRequest>("/file1", RootNodeId);
        UNIT_ASSERT_NO_EXCEPTION(future.Wait(WaitTimeout));
        bootstrap.Stop();

        create.SetValue(TErrorResponse(E_FS_EXIST, ""));
        UNIT_ASSERT_EXCEPTION(future.GetValue(), yexception);
    }

    Y_UNIT_TEST(ShouldHandleReadRequest)
    {
        TBootstrap bootstrap;

        const ui64 nodeId = 123;
        const ui64 handleId = 456;
        const ui64 size = 789;

        bootstrap.Service->ReadDataHandler = [&] (auto callContext, auto request) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            UNIT_ASSERT_VALUES_EQUAL(request->GetHandle(), handleId);

            NProto::TReadDataResponse result;
            result.MutableBuffer()->assign(TString(request->GetLength(), 'a'));

            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto read = bootstrap.Fuse->SendRequest<TReadRequest>(
            nodeId, handleId, 0, size);

        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(read.GetValue(), size);

        read = bootstrap.Fuse->SendRequest<TReadRequest>(
            nodeId, handleId, 0, 10_MB);

        UNIT_ASSERT_EXCEPTION(read.GetValueSync(), yexception);
    }

    Y_UNIT_TEST(ShouldUpdateFilesystemMetricsWithInFlight)
    {
        TBootstrap bootstrap;

        auto execute = NewPromise<void>();
        auto requestPromise = NewPromise<NProto::TCreateHandleResponse>();
        bootstrap.Service->CreateHandleHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            execute.SetValue();
            return requestPromise.GetFuture();
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TCreateHandleRequest>("/file1", RootNodeId);

        UNIT_ASSERT(execute.GetFuture().Wait(WaitTimeout));
        bootstrap.StatsRegistry->UpdateStats(false);

        auto counters = bootstrap.Counters
            ->FindSubgroup("component", "fs_ut_fs")
            ->FindSubgroup("host", "cluster")
            ->FindSubgroup("filesystem", FileSystemId)
            ->FindSubgroup("client", "")
            ->FindSubgroup("cloud", "")
            ->FindSubgroup("folder", "")
            ->FindSubgroup("request", "CreateHandle");

        UNIT_ASSERT_VALUES_EQUAL(1, counters->GetCounter("InProgress")->GetAtomic());

        requestPromise.SetValue(TErrorResponse(E_FS_EXIST, ""));
        UNIT_ASSERT_EXCEPTION(future.GetValueSync(), yexception);
    }

    Y_UNIT_TEST(SendInterruptRequestBeforeOriginal)
    {
        TBootstrap bootstrap;
        const ui64 nodeId = 123;
        const ui64 handleId = 456;
        bootstrap.Service->CreateHandleHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(request);
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            NProto::TCreateHandleResponse result;
            result.SetHandle(handleId);
            result.MutableNodeAttr()->SetId(nodeId);
            result.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
            return MakeFuture(result);
        };

        bootstrap.Start();
        bootstrap.InterruptNextRequest();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto create = bootstrap.Fuse->SendRequest<TCreateHandleRequest>("/file1", RootNodeId);
        UNIT_ASSERT_VALUES_EQUAL(create.GetValueSync(), handleId); // no interrupted
    }

    Y_UNIT_TEST(ShouldNotTriggerFatalErrorForCancelledRequests)
    {
        TBootstrap bootstrap;
        auto promise = NewPromise<NProto::TAcquireLockResponse>();
        auto handlerCalled = NewPromise<void>();

        bootstrap.Service->AcquireLockHandler = [&] (auto callContext, auto request) {
            Y_UNUSED(callContext);
            Y_UNUSED(request);
            handlerCalled.TrySetValue();
            return promise.GetFuture();
        };

        bootstrap.StopTriggered.GetFuture().Subscribe([&] (const auto&) {
            promise.SetValue(
                TErrorResponse(E_FS_WOULDBLOCK, "waiting"));
        });

        bootstrap.Start();

        auto future =
            bootstrap.Fuse->SendRequest<TAcquireLockRequest>(0, F_RDLCK);

        handlerCalled.GetFuture().Wait();

        bootstrap.Stop(); // wait till all requests are done writing their stats

        auto counters = bootstrap.Counters
            ->FindSubgroup("component", "fs_ut")
            ->FindSubgroup("request", "AcquireLock");
        UNIT_ASSERT_VALUES_EQUAL(1, counters->GetCounter("Errors")->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, counters->GetCounter("Errors/Fatal")->GetAtomic());
    }

    Y_UNIT_TEST(ShouldNotTriggerFatalErrorsForNewRequestsDuringFuseStop)
    {
        TBootstrap bootstrap;
        auto promise = NewPromise<NProto::TAcquireLockResponse>();
        auto handlerCalled = NewPromise<void>();

        bootstrap.Service->AcquireLockHandler = [&] (auto, auto) {
            handlerCalled.TrySetValue();
            return promise.GetFuture();
        };

        bootstrap.Service->CreateHandleHandler = [&] (auto , auto) {
            UNIT_ASSERT_C(false, "Handler should not be called");
            return MakeFuture(NProto::TCreateHandleResponse{});
        };

        bootstrap.StopTriggered.GetFuture().Subscribe([&] (const auto&) {
            // Make synchronous call. Since Stop is triggered we expect
            // that request will be cancelled and future will contain exception
            auto future = bootstrap.Fuse->SendRequest<TCreateHandleRequest>(
                    "/file1",
                    RootNodeId);
            UNIT_ASSERT_EXCEPTION(future.GetValueSync(), yexception);

            // Let completion queue complete all pending requests.
            promise.SetValue(NProto::TAcquireLockResponse{});
        });

        bootstrap.Start();

        auto future =
            bootstrap.Fuse->SendRequest<TAcquireLockRequest>(0, F_RDLCK);

        // Wait for lock request to reach lock handler in service.
        // Now StopAsync in bootstrap cannot complete immediately.
        // All new requests (CreateHandle) should be rejected immediately.
        handlerCalled.GetFuture().Wait();

        bootstrap.Stop();

        auto counters = bootstrap.Counters
            ->FindSubgroup("component", "fs_ut")
            ->FindSubgroup("request", "CreateHandle");
        UNIT_ASSERT_VALUES_EQUAL(1, counters->GetCounter("Errors")->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, counters->GetCounter("Errors/Fatal")->GetAtomic());
    }

    Y_UNIT_TEST(ShouldProcessDestroyHandleRequestsAsynchronously)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncDestroyHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features);

        const ui64 handle1 = 2;
        const ui64 nodeId1 = 10;
        const ui64 handle2 = 5;
        const ui64 nodeId2 = 11;
        std::atomic_bool releaseFinished = false;
        std::atomic_uint handlerCalled = 0;
        auto counters = bootstrap.Counters->FindSubgroup("component", "fs_ut")
                            ->FindSubgroup("request", "DestroyHandle");
        auto responsePromise1 = NewPromise<NProto::TDestroyHandleResponse>();
        auto responsePromise2 = NewPromise<NProto::TDestroyHandleResponse>();
        bootstrap.Service->SetHandlerDestroyHandle(
            [&,
             responsePromise1,
             responsePromise2](auto callContext, auto request) mutable
            {
                UNIT_ASSERT_VALUES_EQUAL(
                    1,
                    AtomicGet(counters->GetCounter("InProgress")->GetAtomic()));

                UNIT_ASSERT_VALUES_EQUAL(
                    FileSystemId,
                    callContext->FileSystemId);

                if (++handlerCalled == 1) {
                    UNIT_ASSERT(releaseFinished);
                    UNIT_ASSERT_VALUES_EQUAL(handle1, request->GetHandle());
                    UNIT_ASSERT_VALUES_EQUAL(nodeId1, request->GetNodeId());
                    return responsePromise1;
                }
                if (handlerCalled == 2) {
                    UNIT_ASSERT(releaseFinished);
                    UNIT_ASSERT_VALUES_EQUAL(handle2, request->GetHandle());
                    UNIT_ASSERT_VALUES_EQUAL(nodeId2, request->GetNodeId());
                    return responsePromise2;
                }

                UNIT_ASSERT_C(
                    false,
                    "Handler should not be called more than two times");
                return NewPromise<NProto::TDestroyHandleResponse>();
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
            nodeId1,
            handle1,
            O_RDONLY);
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
        future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
            nodeId2,
            handle2,
            O_RDONLY);
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
        releaseFinished = true;

        scheduler->RunAllScheduledTasks();
        responsePromise1.SetValue(NProto::TDestroyHandleResponse{});

        scheduler->RunAllScheduledTasks();
        responsePromise2.SetValue(NProto::TDestroyHandleResponse{});

        UNIT_ASSERT_VALUES_EQUAL(2U, handlerCalled.load());
        UNIT_ASSERT_VALUES_EQUAL(
            0,
            AtomicGet(counters->GetCounter("InProgress")->GetAtomic()));
    }

    Y_UNIT_TEST(ShouldDrainHandleOpsQueueBackToBack)
    {
        // AsyncHandleOperationDrainPeriod is 0, so a non-empty queue is drained
        // back-to-back: every entry is rescheduled with zero delay.
        // With a frozen clock, running only the tasks due "now" fires just the
        // zero-delay tasks, so the whole queue must drain without advancing
        // time. A non-zero period would leave every processing task in the
        // future and this drain would never complete.
        constexpr ui32 requestCount = 3;

        NProto::TFileStoreFeatures features;
        features.SetAsyncDestroyHandleEnabled(true);
        features.SetAsyncHandleOperationIdlePeriod(50);
        features.SetAsyncHandleOperationDrainPeriod(0);

        auto timer = std::make_shared<TTestTimer>();
        auto scheduler = std::make_shared<TTestScheduler>(timer->Now());
        TBootstrap bootstrap(timer, scheduler, features);

        std::atomic_uint handlerCalled = 0;
        bootstrap.Service->SetHandlerDestroyHandle(
            [&](auto, auto)
            {
                ++handlerCalled;
                return MakeFuture(NProto::TDestroyHandleResponse{});
            });

        auto inProgress =
            bootstrap.Counters->FindSubgroup("component", "fs_ut")
                ->FindSubgroup("request", "DestroyHandle")
                ->GetCounter("InProgress");

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        for (ui32 i = 0; i < requestCount; ++i) {
            auto future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
                10 + i,
                2 + i,
                O_RDONLY);
            UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
        }

        UNIT_ASSERT(WaitForCondition(
            WaitTimeout,
            [&]
            {
                scheduler->RunAllScheduledTasksUntilNow();
                return handlerCalled.load() == requestCount
                    && AtomicGet(inProgress->GetAtomic()) == 0;
            }));
    }

    Y_UNIT_TEST(ShouldUseIdlePeriodWhenHandleOpsQueueEmpty)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncDestroyHandleEnabled(true);
        features.SetAsyncHandleOperationIdlePeriod(50);
        features.SetAsyncHandleOperationDrainPeriod(0);
        const auto idlePeriod = TDuration::MilliSeconds(
            features.GetAsyncHandleOperationIdlePeriod());

        auto timer = std::make_shared<TTestTimer>();
        auto scheduler = std::make_shared<TTestScheduler>(timer->Now());
        TBootstrap bootstrap(timer, scheduler, features);

        std::atomic_uint handlerCalled = 0;
        bootstrap.Service->SetHandlerDestroyHandle(
            [&](auto, auto)
            {
                ++handlerCalled;
                return MakeFuture(NProto::TDestroyHandleResponse{});
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        scheduler->RunAllScheduledTasksUntilNow();

        auto future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
            10,
            2,
            O_RDONLY);
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));

        // The only scheduled poll is idlePeriod in the future, so at the
        // current (frozen) time it does not fire and the entry stays pending.
        scheduler->RunAllScheduledTasksUntilNow();
        UNIT_ASSERT_VALUES_EQUAL(0U, handlerCalled.load());

        // Once the backoff elapses the poll fires and picks up the entry.
        timer->AdvanceTime(idlePeriod);
        scheduler->AdvanceTime(idlePeriod);
        UNIT_ASSERT(WaitForCondition(
            WaitTimeout,
            [&]
            {
                scheduler->RunAllScheduledTasksUntilNow();
                return handlerCalled.load() == 1;
            }));
    }

    Y_UNIT_TEST(ShouldProcessReadOnlyDestroyHandleRequestsAsynchronously)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncDestroyReadOnlyHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features);

        const ui64 readOnlyHandle = 2;
        const ui64 readOnlyNodeId = 10;
        const ui64 writeHandle = 5;
        const ui64 writeNodeId = 11;
        std::atomic_bool readOnlyReleaseFinished = false;
        std::atomic_bool writeReleaseFinished = false;
        std::atomic_uint handlerCalled = 0;
        auto counters = bootstrap.Counters->FindSubgroup("component", "fs_ut")
                            ->FindSubgroup("request", "DestroyHandle");
        auto readOnlyResponsePromise =
            NewPromise<NProto::TDestroyHandleResponse>();
        auto writeResponsePromise =
            NewPromise<NProto::TDestroyHandleResponse>();
        bootstrap.Service->SetHandlerDestroyHandle(
            [&,
             readOnlyResponsePromise,
             writeResponsePromise](auto callContext, auto request) mutable
            {
                UNIT_ASSERT_VALUES_EQUAL(
                    1,
                    AtomicGet(counters->GetCounter("InProgress")->GetAtomic()));

                UNIT_ASSERT_VALUES_EQUAL(
                    FileSystemId,
                    callContext->FileSystemId);

                ++handlerCalled;
                if (request->GetHandle() == readOnlyHandle) {
                    UNIT_ASSERT(readOnlyReleaseFinished);
                    UNIT_ASSERT_VALUES_EQUAL(
                        readOnlyNodeId,
                        request->GetNodeId());
                    return readOnlyResponsePromise;
                }

                if (request->GetHandle() == writeHandle) {
                    UNIT_ASSERT(!writeReleaseFinished);
                    UNIT_ASSERT_VALUES_EQUAL(writeNodeId, request->GetNodeId());
                    return writeResponsePromise;
                }

                UNIT_ASSERT_C(false, "Unexpected DestroyHandle request");
                return NewPromise<NProto::TDestroyHandleResponse>();
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
            readOnlyNodeId,
            readOnlyHandle,
            O_RDONLY);
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
        readOnlyReleaseFinished = true;
        UNIT_ASSERT_VALUES_EQUAL(0U, handlerCalled.load());
        UNIT_ASSERT_VALUES_EQUAL(
            0,
            AtomicGet(counters->GetCounter("InProgress")->GetAtomic()));

        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(1U, handlerCalled.load());
        UNIT_ASSERT_VALUES_EQUAL(
            1,
            AtomicGet(counters->GetCounter("InProgress")->GetAtomic()));
        readOnlyResponsePromise.SetValue(NProto::TDestroyHandleResponse{});

        future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
            writeNodeId,
            writeHandle,
            O_WRONLY);
        UNIT_ASSERT_EXCEPTION(
            future.GetValue(ExceptionWaitTimeout),
            yexception);
        UNIT_ASSERT_VALUES_EQUAL(2U, handlerCalled.load());
        UNIT_ASSERT_VALUES_EQUAL(
            1,
            AtomicGet(counters->GetCounter("InProgress")->GetAtomic()));

        writeResponsePromise.SetValue(NProto::TDestroyHandleResponse{});
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
        writeReleaseFinished = true;
        UNIT_ASSERT_VALUES_EQUAL(
            0,
            AtomicGet(counters->GetCounter("InProgress")->GetAtomic()));
    }

    Y_UNIT_TEST(ShouldNotSetAsyncCreateHandleIfFeatureDisabled)
    {
        // The vhost-side feature gate is off, so the client must not advertise
        // async create support even for an otherwise eligible read-only open.
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        std::atomic_uint confirmCalled = 0;

        bootstrap.Service->SetHandlerCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT(!request->GetAllowAsyncCreateHandle());

                NProto::TCreateHandleResponse response;
                response.SetHandle(handle);
                response.MutableNodeAttr()->SetId(nodeId);
                response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerConfirmCreateHandle(
            [&](auto, auto)
            {
                ++confirmCalled;
                return MakeFuture(NProto::TConfirmCreateHandleResponse{});
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId);
        UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout));

        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load());
    }

    Y_UNIT_TEST(ShouldNotSetAsyncCreateHandleIfAsyncDestroyDisabled)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncCreateHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        std::atomic_uint confirmCalled = 0;

        bootstrap.Service->SetHandlerCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT(!request->GetAllowAsyncCreateHandle());

                NProto::TCreateHandleResponse response;
                response.SetHandle(handle);
                response.MutableNodeAttr()->SetId(nodeId);
                response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerConfirmCreateHandle(
            [&](auto, auto)
            {
                ++confirmCalled;
                return MakeFuture(NProto::TConfirmCreateHandleResponse{});
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId);
        UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout));

        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load());
    }

    Y_UNIT_TEST(ShouldSynchronouslyConfirmUnexpectedAsyncCreateHandleResponse)
    {
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        ui64 requestId = 0;
        ui32 flags = 0;
        TString fileSystemId;
        TString headers;
        std::atomic_uint confirmCalled = 0;
        auto confirmPromise =
            NewPromise<NProto::TConfirmCreateHandleResponse>();

        bootstrap.Service->SetHandlerCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(
                    FileSystemId,
                    callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT(!request->GetAllowAsyncCreateHandle());

                requestId = request->GetHeaders().GetRequestId()
                    ? request->GetHeaders().GetRequestId()
                    : callContext->RequestId;
                flags = request->GetFlags();
                fileSystemId = request->GetFileSystemId();
                headers = request->GetHeaders().SerializeAsString();

                NProto::TCreateHandleResponse response;
                response.SetHandle(handle);
                response.SetHandleCreatedAsync(true);
                response.MutableNodeAttr()->SetId(nodeId);
                response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerConfirmCreateHandle(
            [&, confirmPromise](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(
                    FileSystemId,
                    callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(
                    fileSystemId,
                    request->GetFileSystemId());
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle());
                UNIT_ASSERT_VALUES_EQUAL(flags, request->GetFlags());
                UNIT_ASSERT_VALUES_EQUAL(
                    requestId,
                    request->GetOriginalRequestId());
                UNIT_ASSERT_VALUES_EQUAL(
                    headers,
                    request->GetHeaders().SerializeAsString());
                ++confirmCalled;
                return confirmPromise;
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId);
        UNIT_ASSERT_EXCEPTION(
            future.GetValue(ExceptionWaitTimeout),
            yexception);
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());

        auto errorCounter =
            bootstrap.Counters->GetSubgroup("component", "fs_ut")
                ->GetCounter(
                    "AppCriticalEvents/UnexpectedAsyncCreateHandleResponse",
                    true);
        UNIT_ASSERT_VALUES_EQUAL(1, static_cast<int>(*errorCounter));

        confirmPromise.SetValue(NProto::TConfirmCreateHandleResponse{});
        UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout));

        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());
    }

    void CheckIneligibleRequestDoesNotUseAsyncCreate(
        bool create,
        int systemFlags,
        ui32 expectedProtoFlag)
    {
        // Async create is allowed only for plain read-only opens. Create,
        // write, and truncate requests must stay on the synchronous path.
        NProto::TFileStoreFeatures features;
        features.SetAsyncCreateHandleEnabled(true);
        features.SetAsyncDestroyReadOnlyHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        std::atomic_uint confirmCalled = 0;

        bootstrap.Service->SetHandlerCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT(request->GetFlags() & expectedProtoFlag);
                UNIT_ASSERT(!request->GetAllowAsyncCreateHandle());

                NProto::TCreateHandleResponse response;
                response.SetHandle(handle);
                response.MutableNodeAttr()->SetId(nodeId);
                response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerConfirmCreateHandle(
            [&](auto, auto)
            {
                ++confirmCalled;
                return MakeFuture(NProto::TConfirmCreateHandleResponse{});
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        TFuture<ui64> future;
        if (create) {
            auto request =
                std::make_shared<TCreateHandleRequest>("/file1", nodeId);
            request->In->Body.flags |= systemFlags;
            future = bootstrap.Fuse->SendRequest<TCreateHandleRequest>(request);
        } else {
            auto request = std::make_shared<TOpenHandleRequest>(nodeId);
            request->In->Body.flags |= systemFlags;
            future = bootstrap.Fuse->SendRequest<TOpenHandleRequest>(request);
        }

        UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout));

        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load());
    }

    Y_UNIT_TEST(ShouldNotSetAsyncCreateHandleForIneligibleRequests)
    {
        CheckIneligibleRequestDoesNotUseAsyncCreate(
            false,  // open
            O_WRONLY,
            ProtoFlag(NProto::TCreateHandleRequest::E_WRITE));
        CheckIneligibleRequestDoesNotUseAsyncCreate(
            false,  // open
            O_RDWR,
            ProtoFlag(NProto::TCreateHandleRequest::E_WRITE));
        CheckIneligibleRequestDoesNotUseAsyncCreate(
            false,  // open
            O_APPEND,
            ProtoFlag(NProto::TCreateHandleRequest::E_APPEND));
        CheckIneligibleRequestDoesNotUseAsyncCreate(
            false,  // open
            O_TRUNC,
            ProtoFlag(NProto::TCreateHandleRequest::E_TRUNCATE));
        CheckIneligibleRequestDoesNotUseAsyncCreate(
            true,   // create
            O_CREAT,
            ProtoFlag(NProto::TCreateHandleRequest::E_CREATE));
    }

    Y_UNIT_TEST(ShouldNotQueueAsyncCreateHandleForPersistedResponse)
    {
        // The request advertises async support, but the server can still return
        // a persisted handle. In that case there is nothing to confirm later.
        NProto::TFileStoreFeatures features;
        features.SetAsyncCreateHandleEnabled(true);
        features.SetAsyncDestroyReadOnlyHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        std::atomic_uint confirmCalled = 0;

        bootstrap.Service->SetHandlerCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT(request->GetAllowAsyncCreateHandle());

                NProto::TCreateHandleResponse response;
                response.SetHandle(handle);
                response.MutableNodeAttr()->SetId(nodeId);
                response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerConfirmCreateHandle(
            [&](auto, auto)
            {
                ++confirmCalled;
                return MakeFuture(NProto::TConfirmCreateHandleResponse{});
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId);
        UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout));

        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load());
    }

    Y_UNIT_TEST(ShouldQueueAsyncCreateHandleBeforeReplyingOpen)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncCreateHandleEnabled(true);
        features.SetAsyncDestroyReadOnlyHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        std::atomic_bool openFinished = false;
        std::atomic_uint confirmCalled = 0;

        bootstrap.Service->SetHandlerCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT(request->GetAllowAsyncCreateHandle());

                NProto::TCreateHandleResponse response;
                response.SetHandle(handle);
                response.SetHandleCreatedAsync(true);
                response.MutableNodeAttr()->SetId(nodeId);
                response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerConfirmCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT(openFinished);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle());
                UNIT_ASSERT_VALUES_EQUAL(
                    SessionId,
                    request->GetHeaders().GetSessionId());
                ++confirmCalled;
                return MakeFuture(NProto::TConfirmCreateHandleResponse{});
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId);
        UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout));
        openFinished = true;
        // ConfirmCreateHandle is driven by scheduled queue processing, not by
        // the foreground open path.
        UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load());

        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());
    }

    Y_UNIT_TEST(ShouldNotRetryAsyncCreateHandleConfirmationOnError)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncCreateHandleEnabled(true);
        features.SetAsyncDestroyReadOnlyHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        std::atomic_bool openFinished = false;
        std::atomic_uint confirmCalled = 0;

        bootstrap.Service->SetHandlerCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT(request->GetAllowAsyncCreateHandle());

                NProto::TCreateHandleResponse response;
                response.SetHandle(handle);
                response.SetHandleCreatedAsync(true);
                response.MutableNodeAttr()->SetId(nodeId);
                response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerConfirmCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT(openFinished);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle());
                UNIT_ASSERT_VALUES_EQUAL(
                    SessionId,
                    request->GetHeaders().GetSessionId());

                ++confirmCalled;
                NProto::TConfirmCreateHandleResponse response =
                    TErrorResponse(E_REJECTED, "retriable");
                return MakeFuture(response);
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId);
        UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout));
        openFinished = true;

        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());

        auto errorCounter =
            bootstrap.Counters->GetSubgroup("component", "fs_ut")
                ->GetCounter(
                    "AppCriticalEvents/HandleOpsQueueProcessError",
                    true);
        UNIT_ASSERT_VALUES_EQUAL(1, static_cast<int>(*errorCounter));

        // Retriable errors are handled by DurableClient and must not reach this
        // layer in production. Any error returned here is final.
        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());
    }

    Y_UNIT_TEST(ShouldConfirmAsyncCreateHandleSynchronouslyIfQueueOverflows)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncCreateHandleEnabled(true);
        features.SetAsyncDestroyReadOnlyHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        // Keep the ring buffer smaller than a serialized queued create entry so
        // AddCreateRequest returns QueueOverflow.
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features, 20);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        std::atomic_uint confirmCalled = 0;
        auto confirmPromise = NewPromise<NProto::TConfirmCreateHandleResponse>();

        bootstrap.Service->SetHandlerCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT(request->GetAllowAsyncCreateHandle());

                NProto::TCreateHandleResponse response;
                response.SetHandle(handle);
                response.SetHandleCreatedAsync(true);
                response.MutableNodeAttr()->SetId(nodeId);
                response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerConfirmCreateHandle(
            [&, confirmPromise](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle());
                UNIT_ASSERT_VALUES_EQUAL(
                    SessionId,
                    request->GetHeaders().GetSessionId());
                ++confirmCalled;
                return confirmPromise;
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId);
        // Queue overflow forces synchronous confirmation before ReplyOpen, so
        // the FUSE request must stay pending while confirmPromise is unresolved.
        UNIT_ASSERT_EXCEPTION(
            future.GetValue(ExceptionWaitTimeout),
            yexception);
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());

        confirmPromise.SetValue(NProto::TConfirmCreateHandleResponse{});
        UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout));

        // The create was not queued, so background queue processing must not
        // issue another ConfirmCreateHandle.
        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());
    }

    Y_UNIT_TEST(ShouldNotRetryOverflowAsyncCreateConfirmationOrCleanupErrors)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncCreateHandleEnabled(true);
        features.SetAsyncDestroyReadOnlyHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        // Keep the ring buffer smaller than a serialized queued create entry so
        // AddCreateRequest returns QueueOverflow.
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features, 20);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        std::atomic_uint confirmCalled = 0;
        std::atomic_uint destroyCalled = 0;

        bootstrap.Service->SetHandlerCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT(request->GetAllowAsyncCreateHandle());

                NProto::TCreateHandleResponse response;
                response.SetHandle(handle);
                response.SetHandleCreatedAsync(true);
                response.MutableNodeAttr()->SetId(nodeId);
                response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerConfirmCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle());
                UNIT_ASSERT_VALUES_EQUAL(
                    SessionId,
                    request->GetHeaders().GetSessionId());

                ++confirmCalled;
                NProto::TConfirmCreateHandleResponse response =
                    TErrorResponse(E_REJECTED, "retriable");
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerDestroyHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle());
                ++destroyCalled;
                NProto::TDestroyHandleResponse response =
                    TErrorResponse(E_REJECTED, "retriable");
                return MakeFuture(response);
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId);
        UNIT_ASSERT_EXCEPTION(future.GetValue(WaitTimeout), yexception);
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());
        UNIT_ASSERT_VALUES_EQUAL(1U, destroyCalled.load());

        auto errorCounter =
            bootstrap.Counters->GetSubgroup("component", "fs_ut")
                ->GetCounter(
                    "AppCriticalEvents/ConfirmCreateHandleFailed",
                    true);
        UNIT_ASSERT_VALUES_EQUAL(1, static_cast<int>(*errorCounter));

        auto cleanupErrorCounter =
            bootstrap.Counters->GetSubgroup("component", "fs_ut")
                ->GetCounter(
                    "AppCriticalEvents/AsyncCreateHandleCleanupFailed",
                    true);
        UNIT_ASSERT_VALUES_EQUAL(
            1,
            static_cast<int>(*cleanupErrorCounter));

        // Retriable errors are handled by the durable client. The synchronous
        // fallback and cleanup must not add their own retry loops.
        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());
        UNIT_ASSERT_VALUES_EQUAL(1U, destroyCalled.load());
    }

    Y_UNIT_TEST(ShouldFailOpenIfOverflowAsyncCreateConfirmationFails)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncCreateHandleEnabled(true);
        features.SetAsyncDestroyReadOnlyHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        // Keep the ring buffer smaller than a serialized queued create entry so
        // AddCreateRequest returns QueueOverflow.
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features, 20);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        std::atomic_uint confirmCalled = 0;
        std::atomic_uint destroyCalled = 0;
        auto confirmPromise = NewPromise<NProto::TConfirmCreateHandleResponse>();
        auto destroyPromise = NewPromise<NProto::TDestroyHandleResponse>();

        bootstrap.Service->SetHandlerCreateHandle(
            [&](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT(request->GetAllowAsyncCreateHandle());

                NProto::TCreateHandleResponse response;
                response.SetHandle(handle);
                response.SetHandleCreatedAsync(true);
                response.MutableNodeAttr()->SetId(nodeId);
                response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
                return MakeFuture(response);
            });

        bootstrap.Service->SetHandlerConfirmCreateHandle(
            [&, confirmPromise](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle());
                UNIT_ASSERT_VALUES_EQUAL(
                    SessionId,
                    request->GetHeaders().GetSessionId());
                ++confirmCalled;
                return confirmPromise;
            });

        bootstrap.Service->SetHandlerDestroyHandle(
            [&, destroyPromise](auto callContext, auto request)
            {
                UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle());
                ++destroyCalled;
                return destroyPromise;
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId);
        UNIT_ASSERT_EXCEPTION(
            future.GetValue(ExceptionWaitTimeout),
            yexception);
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());

        NProto::TConfirmCreateHandleResponse response =
            TErrorResponse(E_FS_NOENT, "final");
        confirmPromise.SetValue(std::move(response));
        UNIT_ASSERT_EXCEPTION(
            future.GetValue(ExceptionWaitTimeout),
            yexception);
        UNIT_ASSERT_VALUES_EQUAL(1U, destroyCalled.load());

        destroyPromise.SetValue(NProto::TDestroyHandleResponse{});
        UNIT_ASSERT_EXCEPTION_CONTAINS(
            future.GetValue(WaitTimeout),
            yexception,
            ToString(-ENOENT));

        // The create was never queued, so background queue processing must not
        // retry the failed synchronous confirmation.
        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());
    }

    Y_UNIT_TEST(ShouldRestoreAndDrainAsyncCreateHandleQueueAfterSessionRestart)
    {
        const TString sessionId = CreateGuidAsString();

        NProto::TFileStoreFeatures features;
        features.SetAsyncCreateHandleEnabled(true);
        features.SetAsyncDestroyReadOnlyHandleEnabled(true);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        std::atomic<ui64> requestId = 0;
        std::atomic_uint confirmCalled = 0;
        auto firstConfirmPromise =
            NewPromise<NProto::TConfirmCreateHandleResponse>();

        auto createBootstrap = [&]()
        {
            TBootstrap bootstrap(
                CreateWallClockTimer(),
                std::make_shared<TTestScheduler>(TInstant::Zero()),
                features);

            bootstrap.Service->CreateSessionHandler =
                [features, sessionId](auto, auto)
            {
                NProto::TCreateSessionResponse result;
                result.MutableSession()->SetSessionId(sessionId);
                result.MutableFileStore()->SetBlockSize(4096);
                result.MutableFileStore()->MutableFeatures()->CopyFrom(
                    features);
                result.MutableFileStore()->SetFileSystemId(FileSystemId);
                return MakeFuture(result);
            };

            return bootstrap;
        };

        {
            auto bootstrap = createBootstrap();

            bootstrap.Service->SetHandlerCreateHandle(
                [&](auto callContext, auto request)
                {
                    UNIT_ASSERT_VALUES_EQUAL(
                        FileSystemId,
                        callContext->FileSystemId);
                    UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                    UNIT_ASSERT(request->GetAllowAsyncCreateHandle());

                    NProto::TCreateHandleResponse response;
                    response.SetHandle(handle);
                    response.SetHandleCreatedAsync(true);
                    response.MutableNodeAttr()->SetId(nodeId);
                    response.MutableNodeAttr()->SetType(
                        NProto::E_REGULAR_NODE);
                    return MakeFuture(response);
                });

            bootstrap.Service->SetHandlerConfirmCreateHandle(
                [&, firstConfirmPromise](auto callContext, auto request)
                {
                    UNIT_ASSERT_VALUES_EQUAL(
                        FileSystemId,
                        callContext->FileSystemId);
                    UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                    UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle());
                    UNIT_ASSERT_VALUES_EQUAL(
                        sessionId,
                        request->GetHeaders().GetSessionId());
                    UNIT_ASSERT(request->GetOriginalRequestId());
                    UNIT_ASSERT_VALUES_EQUAL(
                        ProtoFlag(NProto::TCreateHandleRequest::E_READ),
                        request->GetFlags());
                    requestId = request->GetOriginalRequestId();
                    ++confirmCalled;
                    return firstConfirmPromise;
                });

            bootstrap.Start();
            Y_DEFER {
                bootstrap.Stop();
            };

            auto future =
                bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId);
            UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout));

            auto* scheduler =
                dynamic_cast<TTestScheduler*>(bootstrap.Scheduler.get());
            UNIT_ASSERT(scheduler);
            scheduler->RunAllScheduledTasks();

            // The confirmation is still in flight, so the durable queue entry
            // remains in place for the restarted loop to reload and confirm.
            UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());
            UNIT_ASSERT(requestId.load());

            auto suspend = bootstrap.Loop->SuspendAsync();
            UNIT_ASSERT(WaitForCondition(
                WaitTimeout,
                [&]
                {
                    scheduler->RunAllScheduledTasksUntilNow();
                    return suspend.HasValue() || suspend.HasException();
                }));
            UNIT_ASSERT_NO_EXCEPTION(suspend.GetValue(WaitTimeout));

            // A clean Stop() destroys the session and removes the queue. Drop
            // the loop after suspend to model a vhost restart with the session
            // and queue directory still present.
            bootstrap.Loop = nullptr;
        }

        // Complete the abandoned request after its filesystem has been
        // destroyed. Its weak callback must not modify the durable queue.
        firstConfirmPromise.SetValue(NProto::TConfirmCreateHandleResponse{});

        {
            auto bootstrap = createBootstrap();

            bootstrap.Service->SetHandlerConfirmCreateHandle(
                [&](auto callContext, auto request)
                {
                    UNIT_ASSERT_VALUES_EQUAL(
                        FileSystemId,
                        callContext->FileSystemId);
                    UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());
                    UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle());
                    UNIT_ASSERT_VALUES_EQUAL(
                        sessionId,
                        request->GetHeaders().GetSessionId());
                    UNIT_ASSERT_VALUES_EQUAL(
                        requestId.load(),
                        request->GetOriginalRequestId());
                    UNIT_ASSERT_VALUES_EQUAL(
                        ProtoFlag(NProto::TCreateHandleRequest::E_READ),
                        request->GetFlags());
                    ++confirmCalled;
                    return MakeFuture(NProto::TConfirmCreateHandleResponse{});
                });

            bootstrap.Start();
            Y_DEFER {
                bootstrap.Stop();
            };

            UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());

            auto* scheduler =
                dynamic_cast<TTestScheduler*>(bootstrap.Scheduler.get());
            UNIT_ASSERT(scheduler);
            scheduler->RunAllScheduledTasks();

            UNIT_ASSERT_VALUES_EQUAL(2U, confirmCalled.load());

            // The restarted loop drained and popped the durable entry, so
            // another scheduling pass must not confirm the same handle again.
            scheduler->RunAllScheduledTasks();
            UNIT_ASSERT_VALUES_EQUAL(2U, confirmCalled.load());
        }
    }

    Y_UNIT_TEST(ShouldNotRetryDestroyHandleAndRaiseCritEventOnError)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncDestroyHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features);

        const ui64 handle = 2;
        const ui64 nodeId = 10;
        std::atomic_uint handlerCalled = 0;
        auto responsePromise = NewPromise<NProto::TDestroyHandleResponse>();
        bootstrap.Service->SetHandlerDestroyHandle(
            [&, responsePromise](auto callContext, auto request)
            {
                ++handlerCalled;
                UNIT_ASSERT_VALUES_EQUAL(
                    FileSystemId,
                    callContext->FileSystemId);
                UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle());
                UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId());

                return responsePromise;
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
            nodeId,
            handle,
            O_RDONLY);
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));

        scheduler->RunAllScheduledTasks();
        NProto::TDestroyHandleResponse response =
            TErrorResponse(E_REJECTED, "retriable");
        responsePromise.SetValue(std::move(response));

        auto errorCounter =
            bootstrap.Counters->GetSubgroup("component", "fs_ut")
                ->GetCounter(
                    "AppCriticalEvents/AsyncDestroyHandleFailed",
                    true);

        UNIT_ASSERT_VALUES_EQUAL(1, static_cast<int>(*errorCounter));

        // Retriable errors are handled by DurableClient and must not reach this
        // layer in production. Any error returned here is final.
        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(1U, handlerCalled.load());
    }

    Y_UNIT_TEST(ShouldPostponeDestroyHandleRequestIfHandleOpsQueueOverflows)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncDestroyHandleEnabled(true);
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features, 20);

        const ui64 handle1 = 2;
        const ui64 nodeId1 = 10;
        const ui64 handle2 = 5;
        const ui64 nodeId2 = 11;
        std::atomic_uint handlerCalled = 0;
        auto responsePromise = NewPromise<NProto::TDestroyHandleResponse>();
        bootstrap.Service->SetHandlerDestroyHandle(
            [&, responsePromise](auto callContext, auto request) mutable
            {
                Y_UNUSED(callContext);
                Y_UNUSED(request);

                if (++handlerCalled == 1) {
                    UNIT_ASSERT_VALUES_EQUAL(handle1, request->GetHandle());
                    UNIT_ASSERT_VALUES_EQUAL(nodeId1, request->GetNodeId());
                    return responsePromise;
                }

                if (handlerCalled == 2) {
                    UNIT_ASSERT_VALUES_EQUAL(handle2, request->GetHandle());
                    UNIT_ASSERT_VALUES_EQUAL(nodeId2, request->GetNodeId());
                }

                return NewPromise<NProto::TDestroyHandleResponse>();
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto moduleCounters = bootstrap.GetHandleOpsQueueCounters();
        auto aggregateModuleCounters =
            bootstrap.GetAggregateHandleOpsQueueCounters();
        bootstrap.ModuleStatsRegistry->UpdateStats(true);

        auto entryCount = moduleCounters->FindCounter("EntryCount");
        UNIT_ASSERT(entryCount);
        auto capacityBytes = moduleCounters->FindCounter("CapacityBytes");
        UNIT_ASSERT(capacityBytes);
        auto overflowErrorCount =
            moduleCounters->FindCounter("OverflowErrorCount");
        UNIT_ASSERT(overflowErrorCount);
        auto serializationErrorCount =
            moduleCounters->FindCounter("SerializationErrorCount");
        UNIT_ASSERT(serializationErrorCount);
        auto parseErrorCount =
            moduleCounters->FindCounter("ParseErrorCount");
        UNIT_ASSERT(parseErrorCount);

        auto aggregateOverflowErrorCount =
            aggregateModuleCounters->FindCounter("OverflowErrorCount");
        UNIT_ASSERT(aggregateOverflowErrorCount);
        auto aggregateSerializationErrorCount =
            aggregateModuleCounters->FindCounter("SerializationErrorCount");
        UNIT_ASSERT(aggregateSerializationErrorCount);
        auto aggregateParseErrorCount =
            aggregateModuleCounters->FindCounter("ParseErrorCount");
        UNIT_ASSERT(aggregateParseErrorCount);

        UNIT_ASSERT_VALUES_EQUAL(0, entryCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(20, capacityBytes->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, overflowErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, serializationErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, parseErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, aggregateOverflowErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(
            0,
            aggregateSerializationErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, aggregateParseErrorCount->GetAtomic());

        auto counters = bootstrap.Counters
            ->FindSubgroup("component", "fs_ut")
            ->FindSubgroup("request", "DestroyHandle");
        auto future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
            nodeId1,
            handle1,
            O_RDONLY);
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
        bootstrap.ModuleStatsRegistry->UpdateStats(true);
        UNIT_ASSERT_VALUES_EQUAL(1, entryCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(20, capacityBytes->GetAtomic());

        UNIT_ASSERT_VALUES_EQUAL(
            0,
            AtomicGet(counters->GetCounter("InProgress")->GetAtomic()));

        future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
            nodeId2,
            handle2,
            O_RDONLY);

        // Second request should wait until the first request is processed.
        UNIT_ASSERT_EXCEPTION(
            future.GetValue(ExceptionWaitTimeout),
            yexception);
        bootstrap.ModuleStatsRegistry->UpdateStats(true);
        UNIT_ASSERT_VALUES_EQUAL(1, entryCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(1, overflowErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, serializationErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, parseErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(1, aggregateOverflowErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(
            0,
            aggregateSerializationErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, aggregateParseErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(
            1,
            AtomicGet(counters->GetCounter("InProgress")->GetAtomic()));

        // A final error still pops the first entry, so it must also retry the
        // postponed release.
        scheduler->RunAllScheduledTasks();
        responsePromise.SetValue(TErrorResponse(E_FS_NOENT, "final"));
        UNIT_ASSERT_VALUES_EQUAL(1u, handlerCalled.load());

        // After the first request is removed, the second request should be
        // completed and added to the HandleOpsQueue.
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(
            0,
            AtomicGet(counters->GetCounter("InProgress")->GetAtomic()));

        // Check that second request was added to the queue and processed later.
        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(2u, handlerCalled.load());
    }

    Y_UNIT_TEST(ShouldRetryPostponedReleaseAfterAsyncCreateHandleConfirmation)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncCreateHandleEnabled(true);
        features.SetAsyncDestroyHandleEnabled(true);

        const ui64 nodeId = 10;
        const ui64 handle = 2;
        auto asyncResponse = [=]
        {
            NProto::TCreateHandleResponse response;
            response.SetHandle(handle);
            response.SetHandleCreatedAsync(true);
            response.MutableNodeAttr()->SetId(nodeId);
            response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
            return response;
        };
        NProto::TCreateHandleRequest createRequest;

        {
            auto scheduler = std::make_shared<TTestScheduler>();
            TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features);
            bootstrap.Service->SetHandlerCreateHandle(
                [&](auto, auto request)
                {
                    createRequest = *request;
                    return MakeFuture(asyncResponse());
                });
            bootstrap.Service->SetHandlerConfirmCreateHandle(
                [](auto, auto)
                {
                    return MakeFuture(NProto::TConfirmCreateHandleResponse{});
                });

            bootstrap.Start();
            Y_DEFER {
                bootstrap.Stop();
            };
            UNIT_ASSERT_VALUES_EQUAL(
                handle,
                bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId)
                    .GetValue(WaitTimeout));
            scheduler->RunAllScheduledTasks();
        }

        NProto::TQueueEntry createEntry;
        auto* queued = createEntry.MutableQueuedCreateHandleRequest();
        *queued->MutableRequest() = createRequest;
        queued->SetHandle(handle);
        queued->SetNodeId(nodeId);
        queued->SetOriginalRequestId(
            createRequest.GetHeaders().GetRequestId());

        NProto::TQueueEntry destroyEntry;
        destroyEntry.MutableDestroyHandleRequest()->SetHandle(handle);
        destroyEntry.MutableDestroyHandleRequest()->SetNodeId(nodeId);

        // protobuf entries have an 8-byte header and unaligned data. This fits
        // the create-confirm entry, but leaves less than one destroy entry
        // free.
        const ui32 queueSize = static_cast<ui32>(
            8 + createEntry.ByteSizeLong() +
            (8 + destroyEntry.ByteSizeLong()) / 2);

        // open fills the queue with a create confirmation. Its release
        // overflows into DelayedReleaseQueue.
        auto scheduler = std::make_shared<TTestScheduler>();
        TBootstrap bootstrap(
            CreateWallClockTimer(), scheduler, features, queueSize);
        std::atomic_uint confirmCalled = 0;
        std::atomic_uint destroyCalled = 0;

        bootstrap.Service->SetHandlerCreateHandle(
            [=](auto, auto)
            {
                return MakeFuture(asyncResponse());
            });
        bootstrap.Service->SetHandlerConfirmCreateHandle(
            [&](auto, auto)
            {
                ++confirmCalled;
                return MakeFuture(NProto::TConfirmCreateHandleResponse{});
            });
        bootstrap.Service->SetHandlerDestroyHandle(
            [&](auto, auto)
            {
                ++destroyCalled;
                return MakeFuture(NProto::TDestroyHandleResponse{});
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto moduleCounters = bootstrap.GetHandleOpsQueueCounters();
        auto entryCount = moduleCounters->FindCounter("EntryCount");
        auto overflowErrorCount =
            moduleCounters->FindCounter("OverflowErrorCount");
        UNIT_ASSERT(entryCount);
        UNIT_ASSERT(overflowErrorCount);

        UNIT_ASSERT_VALUES_EQUAL(
            handle,
            bootstrap.Fuse->SendRequest<TOpenHandleRequest>(nodeId)
                .GetValue(WaitTimeout));
        bootstrap.ModuleStatsRegistry->UpdateStats(true);
        UNIT_ASSERT_VALUES_EQUAL(1, entryCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0, overflowErrorCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load());

        auto release = bootstrap.Fuse->SendRequest<TReleaseRequest>(
            nodeId,
            handle,
            O_RDONLY);
        // Destroy does not fit, so ReleaseImpl parks this FUSE request.
        UNIT_ASSERT_EXCEPTION(release.GetValue(ExceptionWaitTimeout), yexception);
        bootstrap.ModuleStatsRegistry->UpdateStats(true);
        UNIT_ASSERT_VALUES_EQUAL(1, entryCount->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(1, overflowErrorCount->GetAtomic());

        // Confirm pops the create entry.
        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load());
        UNIT_ASSERT_NO_EXCEPTION(release.GetValue(WaitTimeout));

        // The retry queued DestroyHandle. it is processed on the next pass.
        scheduler->RunAllScheduledTasks();
        UNIT_ASSERT_VALUES_EQUAL(1U, destroyCalled.load());
    }

    // We want to ensure that the same file cannot be reused for FileRingBuffers
    // in different FileSystemLoop instances at the same time,
    // because it may lead to file corruption.
    Y_UNIT_TEST(ShouldFailStartWithSameSessionIdIfAsyncDestroyEnabled)
    {
        NProto::TFileStoreFeatures features;
        features.SetAsyncDestroyHandleEnabled(true);
        TBootstrap bootstrap1(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);
        TBootstrap bootstrap2(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        auto error = bootstrap1.Start();
        Y_DEFER {
            bootstrap1.Stop();
        };
        UNIT_ASSERT(!HasError(error));

        error = bootstrap2.Start();
        UNIT_ASSERT(HasError(error));

        auto handleOpsQueueError =
            bootstrap2.Counters->GetSubgroup("component", "fs_ut")
                ->GetCounter(
                    "AppCriticalEvents/HandleOpsQueueCreatingOrDeletingError",
                    true);

        UNIT_ASSERT_VALUES_EQUAL(1, static_cast<int>(*handleOpsQueueError));
    }

    Y_UNIT_TEST(ShouldDrainHandleOpsQueueAfterAsyncDestroyDisabled)
    {
        const ui64 queuedNodeId = 10;
        const ui64 queuedHandle = 2;
        const ui64 directNodeId = 11;
        const ui64 directHandle = 3;

        auto createBootstrap =
            [](bool asyncDestroyHandleEnabled, ISchedulerPtr scheduler)
        {
            NProto::TFileStoreFeatures features;
            features.SetAsyncDestroyHandleEnabled(asyncDestroyHandleEnabled);

            return TBootstrap(
                CreateWallClockTimer(),
                std::move(scheduler),
                features);
        };

        {
            auto bootstrap = createBootstrap(true, CreateScheduler());
            auto destroyStarted = NewPromise<void>();
            bootstrap.Service->SetHandlerDestroyHandle(
                [destroyStarted](auto, auto) mutable
                {
                    destroyStarted.TrySetValue();
                    return NewPromise<NProto::TDestroyHandleResponse>();
                });

            bootstrap.Start();
            Y_DEFER
            {
                bootstrap.Stop();
            };

            auto release = bootstrap.Fuse->SendRequest<TReleaseRequest>(
                queuedNodeId,
                queuedHandle,
                O_RDONLY);
            UNIT_ASSERT_NO_EXCEPTION(release.GetValue(WaitTimeout));
            UNIT_ASSERT(destroyStarted.GetFuture().Wait(WaitTimeout));

            bootstrap.ModuleStatsRegistry->UpdateStats(true);
            auto counters = bootstrap.GetHandleOpsQueueCounters();
            auto entryCount = counters->FindCounter("EntryCount");
            UNIT_ASSERT(entryCount);
            UNIT_ASSERT_VALUES_EQUAL(1, entryCount->GetAtomic());

            auto suspend = bootstrap.Loop->SuspendAsync();
            UNIT_ASSERT(suspend.Wait(WaitTimeout));
        }

        {
            auto scheduler = std::make_shared<TTestScheduler>();
            auto bootstrap = createBootstrap(false, scheduler);
            std::atomic_uint destroyHandleCalled = 0;
            bootstrap.Service->SetHandlerDestroyHandle(
                [&](auto, const auto& request)
                {
                    if (++destroyHandleCalled == 1) {
                        UNIT_ASSERT_VALUES_EQUAL(
                            queuedNodeId,
                            request->GetNodeId());
                        UNIT_ASSERT_VALUES_EQUAL(
                            queuedHandle,
                            request->GetHandle());
                    } else {
                        UNIT_ASSERT_VALUES_EQUAL(
                            directNodeId,
                            request->GetNodeId());
                        UNIT_ASSERT_VALUES_EQUAL(
                            directHandle,
                            request->GetHandle());
                    }
                    return MakeFuture(NProto::TDestroyHandleResponse{});
                });

            bootstrap.Start();
            Y_DEFER
            {
                bootstrap.Stop();
            };

            bootstrap.ModuleStatsRegistry->UpdateStats(true);
            auto counters = bootstrap.GetHandleOpsQueueCounters();
            auto entryCount = counters->FindCounter("EntryCount");
            UNIT_ASSERT(entryCount);
            UNIT_ASSERT_VALUES_EQUAL(1, entryCount->GetAtomic());

            scheduler->RunAllScheduledTasks();
            UNIT_ASSERT_VALUES_EQUAL(1, destroyHandleCalled.load());
            bootstrap.ModuleStatsRegistry->UpdateStats(true);
            UNIT_ASSERT_VALUES_EQUAL(0, entryCount->GetAtomic());

            auto release = bootstrap.Fuse->SendRequest<TReleaseRequest>(
                directNodeId,
                directHandle,
                O_RDONLY);
            UNIT_ASSERT_NO_EXCEPTION(release.GetValue(WaitTimeout));

            // Do not run the test scheduler after Release. If the request were
            // added to HandleOpsQueue, DestroyHandle would not be called and
            // EntryCount would be 1.
            UNIT_ASSERT_VALUES_EQUAL(2, destroyHandleCalled.load());
            bootstrap.ModuleStatsRegistry->UpdateStats(true);
            UNIT_ASSERT_VALUES_EQUAL(0, entryCount->GetAtomic());
        }
    }

    Y_UNIT_TEST(ShouldReadAndWriteWithServerWriteBackCacheEnabled)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;
        const ui64 size = 789;

        std::atomic<int> readDataCalled = 0;

        bootstrap.Service->ReadDataHandler = [&] (auto callContext, auto request) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            UNIT_ASSERT_VALUES_EQUAL(request->GetHandle(), handleId);

            readDataCalled++;

            NProto::TReadDataResponse result;
            result.MutableBuffer()->assign(TString(request->GetLength(), 'a'));
            return MakeFuture(result);
        };

        std::atomic<int> writeDataCalled = 0;

        bootstrap.Service->WriteDataHandler = [&] (auto callContext, auto) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            writeDataCalled++;
            NProto::TWriteDataResponse result;
            return MakeFuture(result);
        };

        std::atomic<int> fsyncCalled = 0;

        bootstrap.Service->FsyncHandler = [&] (auto callContext, auto) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            fsyncCalled++;
            return MakeFuture(NProto::TFsyncResponse());
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        // should read empty data
        auto read = bootstrap.Fuse->SendRequest<TReadRequest>(
            nodeId, handleId, 0, size);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(read.GetValue(), size);
        UNIT_ASSERT_VALUES_EQUAL(1, readDataCalled.load());

        // write request without O_DIRECT should go to write cache
        auto reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_WRONLY;
        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        // should not write (flush) data from cache immediately
        UNIT_ASSERT_VALUES_EQUAL(0, writeDataCalled.load());

        // should read data from cache
        read = bootstrap.Fuse->SendRequest<TReadRequest>(
            nodeId, handleId, 0, size);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(read.GetValue(), size);
        // ReadData should not be called since last time
        UNIT_ASSERT_VALUES_EQUAL(1, readDataCalled.load());

        auto fsync = bootstrap.Fuse->SendRequest<TFsyncRequest>(
            nodeId, handleId, false /* datasync */);
        UNIT_ASSERT_NO_EXCEPTION(fsync.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, fsyncCalled.load());
        // cache should be flushed
        UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());

        read = bootstrap.Fuse->SendRequest<TReadRequest>(
            nodeId, handleId, 0, size);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(read.GetValue(), size);
        // should read data from underlying session after flush
        UNIT_ASSERT_VALUES_EQUAL(2, readDataCalled.load());

        // subsequent fsync does nothing
        fsync = bootstrap.Fuse->SendRequest<TFsyncRequest>(
            nodeId, handleId, false /* datasync */);
        UNIT_ASSERT_NO_EXCEPTION(fsync.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(2, fsyncCalled.load());
        // no new writes (flushes) are expected
        UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());

        // write request without O_DIRECT should go to write cache
        reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_WRONLY;
        write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());

        // call fsync with |datasync == true| - should work the same way it did
        // before
        fsync = bootstrap.Fuse->SendRequest<TFsyncRequest>(
            nodeId, handleId, true /* datasync */);
        UNIT_ASSERT_NO_EXCEPTION(fsync.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(3, fsyncCalled.load());
        // cache should be flushed
        UNIT_ASSERT_VALUES_EQUAL(2, writeDataCalled.load());

        // write request without O_DIRECT should go to write cache
        reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_WRONLY;
        write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(2, writeDataCalled.load());

        // same checks for Flush
        auto flush = bootstrap.Fuse->SendRequest<TFlushRequest>(
            nodeId, handleId);
        UNIT_ASSERT_NO_EXCEPTION(flush.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(3, writeDataCalled.load());
    }

    Y_UNIT_TEST(ShouldFsyncDirWithServerWriteBackCacheEnabled)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;

        std::atomic<int> writeDataCalled = 0;

        bootstrap.Service->WriteDataHandler = [&] (auto callContext, auto) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            writeDataCalled++;
            NProto::TWriteDataResponse result;
            return MakeFuture(result);
        };

        std::atomic<int> fsyncDirCalled = 0;

        bootstrap.Service->FsyncDirHandler = [&] (auto callContext, auto) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            fsyncDirCalled++;
            return MakeFuture(NProto::TFsyncDirResponse());
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        // write request without O_DIRECT should go to write cache
        auto reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_WRONLY;
        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        // should not write (flush) data from cache immediately
        UNIT_ASSERT_VALUES_EQUAL(0, writeDataCalled.load());

        auto dirHandle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(nodeId);
        UNIT_ASSERT(dirHandle.Wait(WaitTimeout));
        auto dirHandleId = dirHandle.GetValue();

        auto fsyncDir = bootstrap.Fuse->SendRequest<TFsyncDirRequest>(
            nodeId, dirHandleId, false /* datasync */);
        UNIT_ASSERT_NO_EXCEPTION(fsyncDir.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, fsyncDirCalled.load());
        // cache should be flushed
        UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());

        // write request without O_DIRECT should go to write cache
        reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_WRONLY;
        write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        // should not write (flush) data from cache immediately
        UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());

        // call FsyncDir with |datasync == true| - should work the same way it
        // did before
        fsyncDir = bootstrap.Fuse->SendRequest<TFsyncDirRequest>(
            nodeId, dirHandleId, true /* datasync */);
        UNIT_ASSERT_NO_EXCEPTION(fsyncDir.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(2, fsyncDirCalled.load());
        // cache should be flushed
        UNIT_ASSERT_VALUES_EQUAL(2, writeDataCalled.load());
    }

    // TODO: create and use metrics for write cache https://github.com/ydb-platform/nbs/issues/4154
    Y_UNIT_TEST(
        ShouldNotUseServerWriteBackCacheForDirectOrSyncHandles)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        // disable automatic flush to make sure that write cache is not flushed
        const ui32 automaticFlushPeriodMs = 0;
        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features,
            1000,
            automaticFlushPeriodMs);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;

        std::atomic<int> writeDataCalled = 0;
        std::atomic<int> firstFlags = -1;
        std::atomic<int> secondFlags = -1;
        std::atomic<int> thirdFlags = -1;

        bootstrap.Service->WriteDataHandler =
            [&](auto callContext, const auto& request)
        {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            const auto index = writeDataCalled.fetch_add(1);
            if (index == 0) {
                firstFlags.store(request->GetFlags());
            } else if (index == 1) {
                secondFlags.store(request->GetFlags());
            } else if (index == 2) {
                thirdFlags.store(request->GetFlags());
            }
            NProto::TWriteDataResponse result;
            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER
        {
            bootstrap.Stop();
        };

        // write request with O_DIRECT should not go to write cache
        auto reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_DIRECT;
        reqWrite->In->Body.flags |= O_WRONLY;
        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());
        UNIT_ASSERT_VALUES_EQUAL(0, firstFlags.load());

        // write request with O_SYNC should not go to write cache
        reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_SYNC;
        reqWrite->In->Body.flags |= O_WRONLY;
        write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(2, writeDataCalled.load());
        UNIT_ASSERT_VALUES_EQUAL(O_SYNC, secondFlags.load());

        // write request with O_DSYNC should not go to write cache
        reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_DSYNC;
        reqWrite->In->Body.flags |= O_WRONLY;
        write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(3, writeDataCalled.load());
        UNIT_ASSERT_VALUES_EQUAL(O_DSYNC, thirdFlags.load());

        // write request without O_DIRECT should go to write cache
        reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_WRONLY;
        write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(3, writeDataCalled.load());
    }

    Y_UNIT_TEST(ShouldFlushWithServerWriteBackCacheEnabled)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;

        std::atomic<int> writeDataCalled = 0;

        bootstrap.Service->WriteDataHandler = [&] (auto callContext, auto) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            writeDataCalled++;
            NProto::TWriteDataResponse result;
            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto writeReq = std::make_shared<TWriteRequest>(nodeId, handleId, 0, CreateBuffer(4096, 'a'));
        writeReq->In->Body.flags |= O_WRONLY;
        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(writeReq);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        // should not write (flush) data from cache immediately
        UNIT_ASSERT_VALUES_EQUAL(0, writeDataCalled.load());

        // same checks for Flush
        auto flush = bootstrap.Fuse->SendRequest<TFlushRequest>(
            nodeId, handleId);
        UNIT_ASSERT_NO_EXCEPTION(flush.GetValue(WaitTimeout));
        // cache should be flushed
        UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());
    }

    Y_UNIT_TEST(ShouldFlushCachedDataBeforeSyncWrite)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        // disable automatic flush to verify ordering around sync writes
        const ui32 automaticFlushPeriodMs = 0;
        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features,
            1000,
            automaticFlushPeriodMs);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;

        std::atomic<int> writeDataCalled = 0;
        std::atomic<int> firstPayload = 0;
        std::atomic<int> secondPayload = 0;

        bootstrap.Service->WriteDataHandler =
            [&](auto callContext, const auto& request)
        {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);

            const auto payload = static_cast<unsigned char>(
                request->GetBuffer()[request->GetBufferOffset()]);
            const auto index = writeDataCalled.fetch_add(1);
            if (index == 0) {
                firstPayload.store(payload);
            } else if (index == 1) {
                secondPayload.store(payload);
            }

            return MakeFuture(NProto::TWriteDataResponse());
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto writeReq = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        writeReq->In->Body.flags |= O_WRONLY;
        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(writeReq);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(0, writeDataCalled.load());

        writeReq = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'b'));
        writeReq->In->Body.flags |= O_WRONLY;
        writeReq->In->Body.flags |= O_SYNC;
        write = bootstrap.Fuse->SendRequest<TWriteRequest>(writeReq);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));

        UNIT_ASSERT_VALUES_EQUAL(2, writeDataCalled.load());
        UNIT_ASSERT_VALUES_EQUAL(static_cast<int>('a'), firstPayload.load());
        UNIT_ASSERT_VALUES_EQUAL(static_cast<int>('b'), secondPayload.load());
    }

    Y_UNIT_TEST(ShouldReleaseWithServerWriteBackCacheEnabled)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;

        std::atomic<int> writeDataCalled = 0;

        bootstrap.Service->WriteDataHandler = [&] (auto callContext, auto) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            writeDataCalled++;
            NProto::TWriteDataResponse result;
            return MakeFuture(result);
        };

        std::atomic<int> destroyHandleCalled = 0;

        bootstrap.Service->SetHandlerDestroyHandle(
            [&](auto callContext, auto)
            {
                UNIT_ASSERT_VALUES_EQUAL(
                    FileSystemId,
                    callContext->FileSystemId);

                destroyHandleCalled++;

                auto promise = NewPromise<NProto::TDestroyHandleResponse>();
                promise.SetValue({});
                return promise.GetFuture();
            });

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        // write request without O_DIRECT should go to write cache
        auto reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_WRONLY;
        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        // should not write (flush) data from cache immediately
        UNIT_ASSERT_VALUES_EQUAL(0, writeDataCalled.load());

        auto future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
            nodeId,
            handleId,
            O_WRONLY);
        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, destroyHandleCalled.load());
        // cache should be flushed
        UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());
    }

    Y_UNIT_TEST(ShouldFlushAutomaticallyWithServerWriteBackCacheEnabled)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        ui32 automaticFlushPeriodMs = 1;
        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features,
            0,
            automaticFlushPeriodMs);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;

        TPromise<void> writeDataCalled = NewPromise();

        bootstrap.Service->WriteDataHandler = [&] (auto callContext, auto) {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            writeDataCalled.SetValue();
            NProto::TWriteDataResponse result;
            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        // write request without O_DIRECT should go to write cache
        auto reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_WRONLY;
        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));

        UNIT_ASSERT(writeDataCalled.GetFuture().Wait(WaitTimeout));
    }

    // We want to ensure that the same file cannot be reused for FileRingBuffers
    // in different FileSystemLoop instances at the same time,
    // because it may lead to file corruption.
    Y_UNIT_TEST(ShouldFailStartWithSameSessionIdIfWriteBackCacheEnabled)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);
        TBootstrap bootstrap1(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);
        TBootstrap bootstrap2(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        auto error = bootstrap1.Start();
        Y_DEFER {
            bootstrap1.Stop();
        };
        UNIT_ASSERT(!HasError(error));

        error = bootstrap2.Start();
        UNIT_ASSERT(HasError(error));

        auto writeBackCacheError =
            bootstrap2.Counters->GetSubgroup("component", "fs_ut")
                ->GetCounter(
                    "AppCriticalEvents/"
                    "WriteBackCacheCreatingOrDeletingError",
                    true);

        UNIT_ASSERT_VALUES_EQUAL(1, static_cast<int>(*writeBackCacheError));
    }

    Y_UNIT_TEST(ShouldNotCrashWhileStoppingWhenForgetRequestIsInFlight)
    {
        TBootstrap bootstrap;

        bootstrap.Start();

        const ui64 nodeId = 123;
        const ui64 refCount = 10;

        auto future = bootstrap.Fuse->SendRequest<TForgetRequest>(
            nodeId,
            refCount);
        bootstrap.Stop();

        // The test is designed to verify the absence of a double-free in a
        // specific scenario. However, it cannot reliably assert success or
        // failure because both execution paths are possible
        try {
            future.Wait(WaitTimeout);
            // Rare case: the request can finish before StopAsync runs
        } catch (...) {
            // Exception happens in two cases:
            // * Most commonly: a request is scheduled, dequeued from the
            // virtio queue and then cancelled
            // * Very rarely: request scheduling is deferred and the FUSE loop
            // has already been destroyed by the time the request reaches the
            // virtio queue, causing a wait timeout in |TFuseVirtioClient|
        }
    }

    Y_UNIT_TEST(ShouldNotHangIfForgetRequestReceivedBeforeInit)
    {
        TBootstrap bootstrap;

        bootstrap.Start(/* sendInitRequest = */ false);

        const ui64 nodeId = 123;
        const ui64 refCount = 10;

        auto forget = bootstrap.Fuse->SendRequest<TForgetRequest>(
            nodeId,
            refCount);
        UNIT_ASSERT_NO_EXCEPTION(forget.GetValue(WaitTimeout));
    }

    Y_UNIT_TEST(ShouldDrainRequestsWhileStopping)
    {
        TBootstrap bootstrap;

        auto writeDataPromise = NewPromise<NProto::TWriteDataResponse>();
        std::atomic<int> writeDataCalled = 0;

        bootstrap.Service->WriteDataHandler = [&](auto, auto) {
            writeDataCalled++;
            return writeDataPromise.GetFuture();
        };

        bootstrap.Service->ReadDataHandler = [&](auto, auto)
        {
            return MakeFuture(NProto::TReadDataResponse());
        };

        bootstrap.Start();

        const ui64 nodeId = 123;
        const ui64 handleId = 456;

        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(
            nodeId, handleId, 0, CreateBuffer(4096, 'a'));

        UNIT_ASSERT(
            WaitForCondition(
                WaitTimeout, [&]() { return writeDataCalled.load(); }));

        auto stopFuture = bootstrap.StopAsync();

        auto read =
            bootstrap.Fuse->SendRequest<TReadRequest>(nodeId, handleId, 0, 10);
        // Request started after StopAsync, expect the request to be cancelled
        UNIT_ASSERT_EXCEPTION(read.GetValue(WaitTimeout), yexception);

        UNIT_ASSERT(!write.HasValue());
        writeDataPromise.SetValue({});
        UNIT_ASSERT(write.GetValueSync());

        UNIT_ASSERT(stopFuture.Wait(WaitTimeout));
    }

    Y_UNIT_TEST(ShouldHandleDoubleInit)
    {
        TBootstrap bootstrap;

        auto writeDataPromise = NewPromise<NProto::TWriteDataResponse>();

        bootstrap.Service->WriteDataHandler = [&](auto, auto) {
            return writeDataPromise.GetFuture();
        };

        bootstrap.Start(/* sendInitRequest = */ false);

        auto init = bootstrap.Fuse->SendRequest<TInitRequest>();
        UNIT_ASSERT_NO_EXCEPTION(init.GetValueSync());

        const ui64 nodeId = 123;
        const ui64 handleId = 456;

        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(
            nodeId, handleId, 0, CreateBuffer(4096, 'a'));

        init = bootstrap.Fuse->SendRequest<TInitRequest>();
        UNIT_ASSERT_NO_EXCEPTION(init.GetValueSync());

        // TODO(svartmetal): it looks like it is better to drain in-flight
        // requests from the previous boot (sent after the first init and
        // before re-init)
        UNIT_ASSERT(!write.HasValue());
        writeDataPromise.SetValue({});
        UNIT_ASSERT(write.GetValueSync());
    }

    Y_UNIT_TEST(ShouldRaiseCritEventWhenErrorWasSentToGuest)
    {
        TBootstrap bootstrap;

        const ui64 nodeId = 123;
        const ui64 handleId = 456;
        const ui64 size = 789;

        bootstrap.Service->ReadDataHandler = [&](auto, auto)
        {
            return MakeFuture(NProto::TReadDataResponse(TErrorResponse(E_IO)));
        };

        bootstrap.Service->WriteDataHandler = [&](auto, auto)
        {
            return MakeFuture(NProto::TWriteDataResponse(TErrorResponse(E_IO)));
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto read = bootstrap.Fuse->SendRequest<TReadRequest>(
            nodeId, handleId, 0, size);

        UNIT_ASSERT_EXCEPTION(read.GetValueSync(), yexception);

        auto errorCounter =
            bootstrap.Counters->GetSubgroup("component", "fs_ut")
                ->GetCounter("AppCriticalEvents/ErrorWasSentToTheGuest", true);

        UNIT_ASSERT_VALUES_EQUAL(1, errorCounter->Val());

        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(
            nodeId, handleId, 0, CreateBuffer(4096, 'a'));

        UNIT_ASSERT_EXCEPTION(write.GetValue(WaitTimeout), yexception);

        UNIT_ASSERT_VALUES_EQUAL(2, errorCounter->Val());
    }

    Y_UNIT_TEST(ShouldSupportWriteBackCacheCapacity4GiB)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features,
            1000,
            1000,
            4_GB + 100500); // writeBackCacheCapacity

        // Narrowing ui64 to ui32 will result in verification failure because
        // 100500 is less than the minimal allowed cache capacity
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };
    }

    void TestShouldSupportZeroCopyWriteByWriteBackCache(
        bool zeroCopyWriteEnabled)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);
        features.SetZeroCopyWriteEnabled(zeroCopyWriteEnabled);

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;

        std::atomic<int> writeDataCalled = 0;

        bootstrap.Service->WriteDataHandler = [&](auto, const auto& request)
        {
            UNIT_ASSERT_VALUES_EQUAL_C(
                zeroCopyWriteEnabled,
                !request->GetIovecs().empty(),
                "Requests generated by TWriteBackCache should use iovecs if "
                "and only if ZeroCopyWriteEnabled feature is on");
            writeDataCalled++;
            NProto::TWriteDataResponse result;
            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER
        {
            bootstrap.Stop();
        };

        // write request without O_DIRECT should go to write cache
        auto reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_WRONLY;
        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT(write.Wait(WaitTimeout));

        auto reqFlush =
            bootstrap.Fuse->SendRequest<TFlushRequest>(nodeId, handleId);

        UNIT_ASSERT(reqFlush.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());
    }

    Y_UNIT_TEST(ShouldSupportZeroCopyWriteByWriteBackCache)
    {
        TestShouldSupportZeroCopyWriteByWriteBackCache(false);
        TestShouldSupportZeroCopyWriteByWriteBackCache(true);
    }

    Y_UNIT_TEST(ShouldFlushAllRequestsBeforeSessionIsDestroyed)
    {
        // The idea is to fill WriteBackCache with requests and to stop session.
        // It should flush both cached and pending requests before session is
        // destroyed. To ensure that there are pending requests, we write more
        // data than the cache capacity (~1_MB) and temporarily prevent write
        // requests from completion.

        // Note: due to the test framework/client limitations, the number of
        // pending requests should not exceed 128 and the size of a message
        // should not exceed 8192 bytes

        constexpr ui64 NodeCount = 8;
        // ThreadPool limitation
        constexpr i64 MaxPendingRequestCount = 4;
        constexpr i64 MaxRequestCount = 1000;
        // Queue buffer limitation
        constexpr ui64 MinByteCount = 7000;
        constexpr ui64 MaxByteCount = 7500;
        constexpr ui64 MaxOffset = 128_KB;
        constexpr TDuration Timeout = TDuration::Seconds(15);

        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        // Disable automatic flush
        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features,
            /* handleOpsQueueSize= */ 1000,
            /* writeBackCacheAutomaticFlushPeriodMs= */ 0,
            WriteBackCacheCapacity);

        auto writeDataPromise = NewPromise();
        std::atomic<int> writeDataCalledCount = 0;

        bootstrap.Service->WriteDataHandler = [&](auto, auto)
        {
            writeDataCalledCount++;
            // The same future cannot be shared between responses
            auto promise = NewPromise<NProto::TWriteDataResponse>();
            writeDataPromise.GetFuture().Subscribe(
                [promise](const auto&) mutable { promise.SetValue({}); });
            return promise;
        };

        bootstrap.Start();
        Y_DEFER
        {
            bootstrap.Stop();
        };

        auto counters = bootstrap.Counters
            ->FindSubgroup("component", "fs_ut_fs")
            ->FindSubgroup("host", "cluster")
            ->FindSubgroup("filesystem", FileSystemId)
            ->FindSubgroup("client", "")
            ->FindSubgroup("cloud", "")
            ->FindSubgroup("folder", "")
            ->FindSubgroup("request", "WriteData");

        auto requestCountSensor = counters->GetCounter("Count");
        auto requestInProgressSensor = counters->GetCounter("InProgress");
        i64 requestCount = 0;

        while (requestInProgressSensor->Val() < MaxPendingRequestCount) {
            UNIT_ASSERT_GT(MaxRequestCount, requestCount);

            ui64 nodeId = RandomNumber(NodeCount) + 123;
            ui64 handleId = nodeId + 456;
            ui64 offset = RandomNumber(MaxOffset);
            ui64 byteCount =
                RandomNumber(MaxByteCount - MinByteCount + 1) + MinByteCount;

            auto reqWrite = std::make_shared<TWriteRequest>(
                nodeId,
                handleId,
                offset,
                CreateBuffer(byteCount, 'a'));

            reqWrite->In->Body.flags |= O_WRONLY;
            bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);

            requestCount++;

            UNIT_ASSERT(WaitForCondition(
                WaitTimeout,
                [&]()
                {
                    const bool requestIsProcessed =
                        requestCount == requestInProgressSensor->Val() +
                                            requestCountSensor->Val();

                    // A pending request will eventually become completed if the
                    // cache is not full. We don't want to count these requests
                    // as pending and will wait instead.

                    // Cache fullness can be detected by non-zero WriteData
                    // attempts from Flush that is triggered when there is no
                    // space to store the request

                    const bool nonZeroPendingIsExpected =
                        requestInProgressSensor->Val() == 0 ||
                        writeDataCalledCount > 0;

                    return requestIsProcessed && nonZeroPendingIsExpected;
                }));
        }

        auto path = TempDir.Path() / "WriteBackCache" / FileSystemId /
                    SessionId / "write_back_cache";

        UNIT_ASSERT(path.Exists());

        auto stopFuture = bootstrap.StopAsync();

        UNIT_ASSERT_VALUES_EQUAL(
            MaxPendingRequestCount,
            requestInProgressSensor->Val());

        // Enable progression of WriteData requests - this will enable flushing
        writeDataPromise.SetValue();
        UNIT_ASSERT(stopFuture.Wait(Timeout));

        UNIT_ASSERT_VALUES_EQUAL(0, requestInProgressSensor->Val());
        UNIT_ASSERT_VALUES_EQUAL(requestCount, requestCountSensor->Val());
        UNIT_ASSERT(!path.Exists());
    }

    void TestShouldHandleZeroCopyReadRequest(bool enableWriteBackCache)
    {
        NProto::TFileStoreFeatures features;
        features.SetZeroCopyReadEnabled(true);
        features.SetServerWriteBackCacheEnabled(enableWriteBackCache);

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;
        const ui64 size = 30;
        const auto data = GenerateValidateData(size, 2);

        bootstrap.Service->ReadDataHandler = [&](auto callContext, auto request)
        {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            UNIT_ASSERT_VALUES_EQUAL(request->GetHandle(), handleId);
            auto& iovecs = request->GetIovecs();
            UNIT_ASSERT_VALUES_EQUAL(1, iovecs.size());
            UNIT_ASSERT_VALUES_EQUAL(request->GetLength(), iovecs[0].GetLength());

            NProto::TReadDataResponse result;
            memcpy(
                reinterpret_cast<void*>(iovecs[0].GetBase()),
                data.data(),
                iovecs[0].GetLength());
            result.SetLength(iovecs[0].GetLength());

            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER
        {
            bootstrap.Stop();
        };

        auto request =
            std::make_shared<TReadRequest>(nodeId, handleId, 0, size);
        auto read = bootstrap.Fuse->SendRequest<TReadRequest>(request);

        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(read.GetValue(), size);
        UNIT_ASSERT_VALUES_EQUAL(
            data,
            TString(reinterpret_cast<char*>(&request->Out->Body), size));
    }

    Y_UNIT_TEST(ShouldHandleZeroCopyReadRequest)
    {
        TestShouldHandleZeroCopyReadRequest(false);
        TestShouldHandleZeroCopyReadRequest(true);
    }

    Y_UNIT_TEST(ShouldHandleZeroCopyReadRequestFallbackToBuffer)
    {
        NProto::TFileStoreFeatures features;
        features.SetZeroCopyReadEnabled(true);

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;
        const ui64 size = 105;
        const auto data = GenerateValidateData(size, 1);

        bootstrap.Service->ReadDataHandler = [&](auto callContext, auto request)
        {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            UNIT_ASSERT_VALUES_EQUAL(request->GetHandle(), handleId);
            auto& iovecs = request->GetIovecs();
            UNIT_ASSERT_VALUES_EQUAL(1, iovecs.size());
            UNIT_ASSERT_VALUES_EQUAL(request->GetLength(), iovecs[0].GetLength());

            NProto::TReadDataResponse result;
            result.MutableBuffer()->assign(data);

            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER
        {
            bootstrap.Stop();
        };

        auto request =
            std::make_shared<TReadRequest>(nodeId, handleId, 0, size);
        auto read = bootstrap.Fuse->SendRequest<TReadRequest>(request);

        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(read.GetValue(), size);
        UNIT_ASSERT_VALUES_EQUAL(
            data,
            TString(reinterpret_cast<char*>(&request->Out->Body), size));
    }

    Y_UNIT_TEST(ShouldEnableZeroCopyReadAndWriteViaLegacyAlias)
    {
        // The legacy ZeroCopyEnabled feature must enable both read and write
        // zero-copy through the FUSE config alias, without the dedicated
        // ZeroCopyReadEnabled / ZeroCopyWriteEnabled features being set.
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);
        features.SetZeroCopyEnabled(true);

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;
        const ui64 size = 30;
        const auto data = GenerateValidateData(size, 2);

        std::atomic<int> readDataCalled = 0;
        bootstrap.Service->ReadDataHandler = [&](auto, auto request)
        {
            UNIT_ASSERT_C(
                !request->GetIovecs().empty(),
                "Legacy ZeroCopyEnabled alias should enable read zero-copy");
            readDataCalled++;
            NProto::TReadDataResponse result;
            memcpy(
                reinterpret_cast<void*>(request->GetIovecs()[0].GetBase()),
                data.data(),
                request->GetIovecs()[0].GetLength());
            result.SetLength(request->GetIovecs()[0].GetLength());
            return MakeFuture(result);
        };

        std::atomic<int> writeDataCalled = 0;
        bootstrap.Service->WriteDataHandler = [&](auto, const auto& request)
        {
            UNIT_ASSERT_C(
                !request->GetIovecs().empty(),
                "Legacy ZeroCopyEnabled alias should enable write zero-copy");
            writeDataCalled++;
            NProto::TWriteDataResponse result;
            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER
        {
            bootstrap.Stop();
        };

        auto readReq =
            std::make_shared<TReadRequest>(nodeId, handleId, 0, size);
        auto read = bootstrap.Fuse->SendRequest<TReadRequest>(readReq);
        UNIT_ASSERT(read.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(read.GetValue(), size);
        UNIT_ASSERT_VALUES_EQUAL(1, readDataCalled.load());

        auto reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_WRONLY;
        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT(write.Wait(WaitTimeout));

        auto reqFlush =
            bootstrap.Fuse->SendRequest<TFlushRequest>(nodeId, handleId);
        UNIT_ASSERT(reqFlush.Wait(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());
    }

    Y_UNIT_TEST(ShouldRestoreAndDrainCacheAfterSessionRestart)
    {
        const TString sessionId = CreateGuidAsString();

        std::atomic<int> writeDataCalled = 0;
        std::atomic<int> writeDataCalled2 = 0;

        const ui64 nodeId = 123;
        const ui64 handleId = 456;

        auto createBootstrap = [&](bool serverWriteBackCacheEnabled,
                                   std::atomic<int>& counter)
        {
            NProto::TFileStoreFeatures features;
            features.SetServerWriteBackCacheEnabled(
                serverWriteBackCacheEnabled);

            TBootstrap bootstrap(
                CreateWallClockTimer(),
                CreateScheduler(),
                features);

            bootstrap.Service->CreateSessionHandler =
                [features, &sessionId](auto, auto)
            {
                NProto::TCreateSessionResponse result;
                result.MutableSession()->SetSessionId(sessionId);
                result.MutableFileStore()->SetBlockSize(4096);
                result.MutableFileStore()->MutableFeatures()->CopyFrom(
                    features);
                result.MutableFileStore()->SetFileSystemId(FileSystemId);
                return MakeFuture(result);
            };

            bootstrap.Service->WriteDataHandler = [&counter](auto, const auto&)
            {
                counter++;
                NProto::TWriteDataResponse result;
                return MakeFuture(result);
            };

            return bootstrap;
        };

        {
            auto bootstrap = createBootstrap(true, writeDataCalled);

            bootstrap.Start();
            Y_DEFER
            {
                bootstrap.Stop();
            };

            auto reqWrite = std::make_shared<TWriteRequest>(
                nodeId,
                handleId,
                0,
                CreateBuffer(4096, 'a'));
            reqWrite->In->Body.flags |= O_WRONLY;
            auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
            UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));

            auto suspend = bootstrap.Loop->SuspendAsync();
            UNIT_ASSERT(suspend.Wait(WaitTimeout));
        }

        // Since write-back cache was enabled, the actual write didn't happen
        // and the request is stored in the persistent queue
        UNIT_ASSERT_VALUES_EQUAL(0, writeDataCalled.load());

        auto path = TempDir.Path() / "WriteBackCache" / FileSystemId /
                    sessionId / "write_back_cache";

        {
            TFileRingBuffer ringBuffer(
                path,
                WriteBackCacheCapacity,
                0,
                EFileRingBufferVersion::V6);

            UNIT_ASSERT(!ringBuffer.Empty());
        }

        {
            auto bootstrap = createBootstrap(false, writeDataCalled2);

            bootstrap.Start();
            Y_DEFER
            {
                bootstrap.Stop();
            };

            // Drain triggers flush immediately
            UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled2.load());

            auto flush =
                bootstrap.Fuse->SendRequest<TFlushRequest>(nodeId, handleId);
            UNIT_ASSERT_NO_EXCEPTION(flush.GetValue(WaitTimeout));

            // cache should be flushed
            UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled2.load());

            auto reqWrite = std::make_shared<TWriteRequest>(
                nodeId,
                handleId,
                0,
                CreateBuffer(4096, 'a'));
            reqWrite->In->Body.flags |= O_WRONLY;
            auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
            UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));

            // Cache is drained and disabled - new requests go directly
            // to the session
            UNIT_ASSERT_VALUES_EQUAL(2, writeDataCalled2.load());

            auto suspend = bootstrap.Loop->SuspendAsync();
            UNIT_ASSERT(suspend.Wait(WaitTimeout));
        }

        {
            TFileRingBuffer ringBuffer(
                path,
                WriteBackCacheCapacity,
                0,
                EFileRingBufferVersion::V6);

            UNIT_ASSERT(ringBuffer.Empty());
        }

        {
            auto bootstrap = createBootstrap(false, writeDataCalled2);

            bootstrap.Start();
            Y_DEFER
            {
                bootstrap.Stop();
            };
        }

        UNIT_ASSERT(!path.Exists());
    }

    Y_UNIT_TEST(ShouldUpdateFileSizeWhenUsingWriteBackCache)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        TMutex mutex;

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        // Set these variables to different before each test case in order to
        // prevent caching node attributes
        std::atomic<ui64> nodeId = 0;

        auto getNodeName = [&]() {
            return Sprintf("file_%lu", nodeId.load());
        };

        std::atomic<ui64> size = 0;

        std::atomic<int> writeDataCalled = 0;
        std::atomic<int> createHandleCalled = 0;
        std::atomic<int> getNodeAttrCalled = 0;
        std::atomic<int> listNodesCalled = 0;
        std::atomic<int> resolvePathCalled = 0;

        // Emulate std::atomic<ui64>::fetch_max - atomically calculate max
        auto fetchMax = [&](ui64 arg)
        {
            ui64 cur = size.load();
            while (cur < arg && !size.compare_exchange_strong(cur, arg)) {
            }
        };

        // The following filestore service requests return node attributes:
        // - CreateHandle
        // - GetNodeAttr
        // - ListNodes
        // - ResolvePath

        auto setAttr = [&](NProto::TNodeAttr* nodeAttr)
        {
            nodeAttr->SetId(nodeId);
            nodeAttr->SetType(NProto::E_REGULAR_NODE);
            nodeAttr->SetSize(size);
        };

        auto proceedServiceOperations = NewPromise();

        auto getProceedServiceOperations = [&]()
        {
            with_lock (mutex) {
                return proceedServiceOperations;
            }
        };

        auto delayedFuture = [&](auto result)
        {
            auto promise = NewPromise<decltype(result)>();
            getProceedServiceOperations().GetFuture().Subscribe(
                [promise, result = std::move(result)](const auto&) mutable
                { promise.SetValue(result); });
            return promise.GetFuture();
        };

        bootstrap.Service->CreateHandleHandler = [&](auto, auto)
        {
            createHandleCalled++;
            NProto::TCreateHandleResponse result;
            result.SetHandle(nodeId + 100);
            setAttr(result.MutableNodeAttr());
            return delayedFuture(result);
        };

        bootstrap.Service->GetNodeAttrHandler = [&](auto, auto)
        {
            getNodeAttrCalled++;
            NProto::TGetNodeAttrResponse result;
            setAttr(result.MutableNode());
            return delayedFuture(result);
        };

        bootstrap.Service->ListNodesHandler = [&](auto, auto)
        {
            listNodesCalled++;
            NProto::TListNodesResponse result;
            result.AddNames()->assign(getNodeName());
            setAttr(result.AddNodes());
            return delayedFuture(result);
        };

        bootstrap.Service->ResolvePathHandler = [&](auto, auto)
        {
            resolvePathCalled++;
            NProto::TResolvePathResponse result;
            result.AddNodeIds(nodeId);
            setAttr(result.MutableAttr());
            return delayedFuture(result);
        };

        // We should also check the amount of bytes returned by ReadData
        // that is affected by the node size

        bootstrap.Service->ReadDataHandler = [&](auto, const auto& rq)
        {
            ui64 end = Min(size.load(), rq->GetOffset() + rq->GetLength());
            ui64 byteCount = end > rq->GetOffset() ? end - rq->GetOffset() : 0;

            NProto::TReadDataResponse result;
            result.MutableBuffer()->assign(TString(byteCount, 'a'));
            return delayedFuture(result);
        };

        // The following Fuse requests call the mentioned above filestore
        // service requests:
        // - FUSE_CREATE -> CreateHandle (creates new node, expected size = 0)
        // - FUSE_OPEN -> CreateHandle (the returned attributes are not used)
        // - FUSE_GETATTR -> GetNodeAttr
        // - FUSE_LOOKUP -> GetNodeAttr
        // - FUSE_READDIR -> ListNodes (the returned attributes are not used)
        // - FUSE_READDIRPLUS -> ListNodes
        // Filestore service request ResolvePath seems to be never called

        // Note: TFileSystem implementation may implicitly cache node attributes
        // returned by the filestore service. We want to ensure that the
        // correct values are cached so it worth to call FUSE_CREATE, FUSE_OPEN
        // and FUSE_READDIR despite the fact that they don't directly return
        // attributes in their responses

        auto getSizeFromGetAttr = [&]()
        {
            auto rq = std::make_shared<TGetAttrRequest>(nodeId);
            return bootstrap.Fuse->SendRequest(rq).Apply(
                [rq](const auto&) { return rq->Out->Body.attr.size; });
        };

        auto getSizeFromOpenHandle = [&]()
        {
            auto rq = std::make_shared<TOpenHandleRequest>(nodeId);
            // The handler for FUSE_OPEN internally calls Session->CreateHandle
            // that returns node attributes that may be cached and returned by
            // the subsequent FUSE_GETATTR call
            return bootstrap.Fuse->SendRequest(rq).Apply(
                [&](const auto&) { return getSizeFromGetAttr(); });
        };

        auto getSizeFromLookup = [&]()
        {
            auto rq = std::make_shared<TLookupRequest>(getNodeName(), nodeId);
            return bootstrap.Fuse->SendRequest(rq).Apply(
                [rq](const auto&) { return rq->Out->Body.attr.size; });
        };

        auto getSizeFromDirectoryRead = [&]()
        {
            auto rq1 = std::make_shared<TOpenDirRequest>(nodeId + 1);
            auto rq2 = std::make_shared<TReadDirRequest>(nodeId + 1, 0);
            auto rq3 = std::make_shared<TReleaseDirRequest>(nodeId + 1, 0);

            return bootstrap.Fuse->SendRequest(rq1)
                .Apply(
                    [&, rq1, rq2](const auto&)
                    {
                        rq2->In->Body.fh = rq1->Out->Body.fh;
                        return bootstrap.Fuse->SendRequest(rq2);
                    })
                .Apply(
                    [&, rq1, rq3](const auto&)
                    {
                        rq3->In->Body.fh = rq1->Out->Body.fh;
                        return bootstrap.Fuse->SendRequest(rq3);
                    })
                .Apply(
                    [&, rq2](const auto&)
                    {
                        auto buf = TStringBuf(
                            reinterpret_cast<const char*>(rq2->Out->Data()),
                            rq2->Out->Header.len - sizeof(rq2->Out->Header));

                        // Skip "." and ".." entries
                        UNIT_ASSERT_LE(
                            sizeof(fuse_direntplus) + dotSize + dotDotSize,
                            buf.size());
                        buf = buf.Skip(dotSize + dotDotSize);

                        const auto* de =
                            reinterpret_cast<const fuse_direntplus*>(
                                buf.data());

                        return de->entry_out.attr.size;
                    });
        };

        auto getSizeFromDirectRead = [&]()
        {
            auto rq = std::make_shared<TReadRequest>(
                nodeId,
                nodeId + 100,
                0,
                1000);
            rq->In->Body.flags |= O_DIRECT;
            return bootstrap.Fuse->SendRequest<TReadRequest>(rq);
        };

        auto getSizeFromCachedRead = [&]()
        {
            auto rq = std::make_shared<TReadRequest>(
                nodeId,
                nodeId + 100,
                0,
                1000);
            return bootstrap.Fuse->SendRequest<TReadRequest>(rq);
        };

        // The following filestore service operations affect node size:
        // - AllocateData
        // - SetNodeAttr (also returns new attributes)
        // - WriteData

        bootstrap.Service->AllocateDataHandler = [&](auto, const auto& rq)
        {
            fetchMax(rq->GetOffset() + rq->GetLength());
            return MakeFuture(NProto::TAllocateDataResponse());
        };

        bootstrap.Service->SetNodeAttrHandler = [&](auto, const auto& rq)
        {
            // Assume that F_SET_ATTR_SIZE is set
            size = rq->GetUpdate().GetSize();
            NProto::TSetNodeAttrResponse result;
            result.MutableNode()->SetId(nodeId);
            result.MutableNode()->SetSize(size);
            return MakeFuture(std::move(result));
        };

        bootstrap.Service->WriteDataHandler = [&](auto, const auto& rq)
        {
            writeDataCalled++;
            fetchMax(
                rq->GetOffset() + rq->GetBuffer().size() -
                rq->GetBufferOffset());
            return MakeFuture(NProto::TWriteDataResponse());
        };

        // We verify reported node size using various request combinations

        bootstrap.Start();
        Y_DEFER
        {
            bootstrap.Stop();
        };

        auto write = [&](ui64 offset, ui64 size, bool direct)
        {
            auto rq = std::make_shared<TWriteRequest>(
                nodeId,
                nodeId + 100,
                offset,
                CreateBuffer(size, 'a'));
            rq->In->Body.flags |= O_WRONLY;
            if (direct) {
                rq->In->Body.flags |= O_DIRECT;
            }
            auto resp = bootstrap.Fuse->SendRequest<TWriteRequest>(rq);
            return resp.GetValue(WaitTimeout) == size;
        };

        auto flush = [&]()
        {
            auto rq = std::make_shared<TFlushRequest>(
                nodeId,
                nodeId + 100);
            auto resp = bootstrap.Fuse->SendRequest<TFlushRequest>(rq);
            resp.Wait(WaitTimeout);
        };

        auto setSize = [&](ui64 size)
        {
            auto rq = std::make_shared<TSetAttrRequest>(nodeId, nodeId + 100);
            rq->In->Body.valid = FUSE_SET_ATTR_SIZE;
            rq->In->Body.size = size;
            UNIT_ASSERT(bootstrap.Fuse->SendRequest(rq).Wait(WaitTimeout));
            return rq->Out->Body.attr.size;
        };

        Y_UNUSED(setSize);

        auto iter = 1;

        auto nextIter = [&]()
        {
            ++iter;
            size = iter * 10;
            nodeId = iter * 5;
            return size.load();
        };

        // Scenario 1:
        // - Request and wait for WriteData with O_DIRECT flag
        // - Get node size
        auto testScenario1 = [&](const TString& testName, auto getSizeFunc)
        {
            auto size = nextIter();
            auto counter = writeDataCalled.load();
            getProceedServiceOperations().TrySetValue();
            UNIT_ASSERT_C(write(size - 5, 9, true), "Scenario 1 " + testName);
            UNIT_ASSERT_VALUES_EQUAL_C(
                counter + 1,
                writeDataCalled.load(),
                "Scenario 1 " + testName);
            UNIT_ASSERT_VALUES_EQUAL_C(
                size + 4,
                getSizeFunc().GetValue(WaitTimeout),
                "Scenario 1 " + testName);
        };

        // Scenario 2:
        // - Request and wait for WriteData without O_DIRECT flag
        // - Get node size
        auto testScenario2 = [&](const TString& testName, auto getSizeFunc)
        {
            auto size = nextIter();
            auto counter = writeDataCalled.load();
            getProceedServiceOperations().TrySetValue();
            UNIT_ASSERT_C(write(size - 5, 9, false), "Scenario 2 " + testName);
            UNIT_ASSERT_VALUES_EQUAL_C(
                counter,
                writeDataCalled.load(),
                "Scenario 2 " + testName);
            UNIT_ASSERT_VALUES_EQUAL_C(
                size + 4,
                getSizeFunc().GetValue(WaitTimeout),
                "Scenario 2 " + testName);
        };

        // Scenario 3:
        // - Request and wait for WriteData without O_DIRECT flag
        // - Initiate a request that returns node size
        // - Request and wait for Flush
        // - Proceed the request that returns node size
        auto testScenario3 = [&](const TString& testName, auto getSizeFunc)
        {
            auto size = nextIter();
            auto counter = writeDataCalled.load();
            with_lock(mutex) {
                proceedServiceOperations = NewPromise();
            }
            UNIT_ASSERT_C(write(size - 5, 9, false), "Scenario 3 " + testName);
            UNIT_ASSERT_VALUES_EQUAL_C(
                counter,
                writeDataCalled.load(),
                "Scenario 3 " + testName);

            auto sizeFuture = getSizeFunc();
            UNIT_ASSERT_C(!sizeFuture.HasValue(), "Scenario 3 " + testName);

            flush();
            UNIT_ASSERT_VALUES_EQUAL_C(
                counter + 1,
                writeDataCalled.load(),
                "Scenario 3 " + testName);

            getProceedServiceOperations().SetValue();
            UNIT_ASSERT_VALUES_EQUAL_C(
                size + 4,
                sizeFuture.GetValue(WaitTimeout),
                "Scenario 3 " + testName);
        };

        auto test = [&](const TString& testName, auto getSizeFunc)
        {
            testScenario1(testName, getSizeFunc);
            testScenario2(testName, getSizeFunc);
            testScenario3(testName, getSizeFunc);
        };

        test("OpenHandle", getSizeFromOpenHandle);
        test("GetAttr", getSizeFromGetAttr);
        test("Lookup", getSizeFromLookup);
        test("DirectoryRead", getSizeFromDirectoryRead);
        test("CachedRead", getSizeFromCachedRead);
        test("DirectRead", getSizeFromDirectRead);
    }

    struct TAttrRaceBootstrap: TBootstrap
    {
        const ui64 ParentNodeId = 101;
        const ui64 NodeId = 1001;
        const ui64 HandleId = 10001;
        std::atomic<ui64> NodeSize = 0;

        TAutoEvent GetNodeAttrRequestEvent;
        TPromise<NProto::TGetNodeAttrResponse> GetNodeAttrResponse;

        TAutoEvent ListNodesRequestEvent;
        TPromise<NProto::TListNodesResponse> ListNodesResponse;

        TPromise<void> WritePromise = NewPromise();
        TPromise<void> AllocatePromise = NewPromise();
        TPromise<void> SetNodeAttrPromise = NewPromise();
        TPromise<void> UnlinkPromise = NewPromise();
        TPromise<void> RenamePromise = NewPromise();

        TAttrRaceBootstrap()
            : TBootstrap(
                CreateWallClockTimer(),
                std::make_shared<TTestScheduler>())
        {
            Service->GetNodeAttrHandler = [&](auto, auto)
            {
                GetNodeAttrRequestEvent.Signal();
                GetNodeAttrResponse = NewPromise<NProto::TGetNodeAttrResponse>();
                return GetNodeAttrResponse.GetFuture();
            };

            Service->ListNodesHandler = [&](auto, auto)
            {
                ListNodesRequestEvent.Signal();
                ListNodesResponse = NewPromise<NProto::TListNodesResponse>();
                return ListNodesResponse.GetFuture();
            };

            Service->WriteDataHandler = [&](auto, const auto& rq)
            {
                NodeSize = Max(
                    NodeSize.load(),
                    rq->GetOffset() + rq->GetBuffer().size());
                NProto::TWriteDataResponse result;
                return MakeFuture(result).Subscribe([&] (auto) {
                    WritePromise.SetValue();
                });
            };

            Service->AllocateDataHandler = [&](auto, const auto& rq)
            {
                NodeSize = Max(
                    NodeSize.load(),
                    rq->GetOffset() + rq->GetLength());
                NProto::TAllocateDataResponse result;
                return MakeFuture(result).Subscribe([&] (auto) {
                    AllocatePromise.SetValue();
                });
            };

            Service->SetNodeAttrHandler = [&](auto, const auto& rq)
            {
                NodeSize = rq->GetUpdate().GetSize();
                NProto::TSetNodeAttrResponse result;
                result.MutableNode()->SetId(NodeId);
                result.MutableNode()->SetSize(NodeSize);
                return MakeFuture(std::move(result)).Subscribe([&] (auto) {
                    SetNodeAttrPromise.SetValue();
                });
            };

            Service->UnlinkNodeHandler = [&](auto, auto)
            {
                NProto::TUnlinkNodeResponse result;
                return MakeFuture(result).Subscribe([&] (auto) {
                    UnlinkPromise.SetValue();
                });
            };

            Service->RenameNodeHandler = [&](auto, auto)
            {
                NProto::TRenameNodeResponse result;
                return MakeFuture(result).Subscribe([&] (auto) {
                    RenamePromise.SetValue();
                });
            };
        }

        void Write(ui64 offset, ui64 size)
        {
            auto rq = std::make_shared<TWriteRequest>(
                NodeId,
                HandleId,
                offset,
                CreateBuffer(size, 'a'));
            rq->In->Body.flags |= O_WRONLY;
            auto resp = Fuse->SendRequest<TWriteRequest>(rq);
            resp.GetValue(WaitTimeout);
            WritePromise.GetFuture().GetValue(WaitTimeout);
        };

        void Allocate(ui64 offset, ui64 size)
        {
            auto rq = std::make_shared<TAllocateRequest>(
                NodeId,
                HandleId,
                offset,
                size);
            auto resp = Fuse->SendRequest<TAllocateRequest>(rq);
            resp.GetValue(WaitTimeout);
            AllocatePromise.GetFuture().GetValue(WaitTimeout);
        };

        void SetSize(ui64 size)
        {
            auto rq = std::make_shared<TSetAttrRequest>(NodeId, HandleId);
            rq->In->Body.valid = FUSE_SET_ATTR_SIZE;
            rq->In->Body.size = size;
            auto resp = Fuse->SendRequest(rq);
            resp.GetValue(WaitTimeout);
            SetNodeAttrPromise.GetFuture().GetValue(WaitTimeout);
        };

        void Unlink(const TString& name)
        {
            auto rq = std::make_shared<TUnlinkRequest>(name, ParentNodeId);
            auto resp = Fuse->SendRequest(rq);
            resp.GetValue(WaitTimeout);
            UnlinkPromise.GetFuture().GetValue(WaitTimeout);
        };

        void RmDir(const TString& name)
        {
            auto rq = std::make_shared<TRmDirRequest>(name, ParentNodeId);
            auto resp = Fuse->SendRequest(rq);
            resp.GetValue(WaitTimeout);
            UnlinkPromise.GetFuture().GetValue(WaitTimeout);
        };

        void Rename(const TString& name, const TString& newName)
        {
            auto rq = std::make_shared<TRenameRequest>(
                name,
                ParentNodeId,
                newName,
                ParentNodeId);
            auto resp = Fuse->SendRequest(rq);
            resp.GetValue(WaitTimeout);
            RenamePromise.GetFuture().GetValue(WaitTimeout);
        };
    };

    Y_UNIT_TEST(
        ShouldNotAllowGuestAttrCachingUponGetAttrAndAttrModificationRace)
    {
        TAttrRaceBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        {
            auto rq = std::make_shared<TGetAttrRequest>(bootstrap.NodeId);
            auto rf = bootstrap.Fuse->SendRequest(rq);
            bootstrap.GetNodeAttrRequestEvent.WaitT(WaitTimeout);
            const ui64 oldSize = bootstrap.NodeSize;
            bootstrap.Write(bootstrap.NodeSize, 4_KB);

            NProto::TGetNodeAttrResponse r;
            auto& n = *r.MutableNode();
            n.SetId(bootstrap.NodeId);
            n.SetSize(oldSize);
            bootstrap.GetNodeAttrResponse.SetValue(std::move(r));
            rf.GetValue(WaitTimeout);

            //
            // Write completed while we were processing GetNodeAttr - the result
            // is stale, shouldn't be cached (attr_timeout should be 0).
            //

            UNIT_ASSERT_VALUES_EQUAL(oldSize, rq->Out->Body.attr.size);
            UNIT_ASSERT_VALUES_EQUAL(0, rq->Out->Body.attr_valid);
            UNIT_ASSERT_VALUES_EQUAL(0, rq->Out->Body.attr_valid_nsec);
        }

        {
            auto rq = std::make_shared<TGetAttrRequest>(bootstrap.NodeId);
            auto rf = bootstrap.Fuse->SendRequest(rq);
            bootstrap.GetNodeAttrRequestEvent.WaitT(WaitTimeout);
            const ui64 oldSize = bootstrap.NodeSize;
            bootstrap.Allocate(bootstrap.NodeSize, 4_KB);

            NProto::TGetNodeAttrResponse r;
            auto& n = *r.MutableNode();
            n.SetId(bootstrap.NodeId);
            n.SetSize(oldSize);
            bootstrap.GetNodeAttrResponse.SetValue(std::move(r));
            rf.GetValue(WaitTimeout);

            //
            // FAllocate completed while we were processing GetNodeAttr - the
            // result is stale, shouldn't be cached (attr_timeout should be 0).
            //

            UNIT_ASSERT_VALUES_EQUAL(oldSize, rq->Out->Body.attr.size);
            UNIT_ASSERT_VALUES_EQUAL(0, rq->Out->Body.attr_valid);
            UNIT_ASSERT_VALUES_EQUAL(0, rq->Out->Body.attr_valid_nsec);
        }

        {
            auto rq = std::make_shared<TGetAttrRequest>(bootstrap.NodeId);
            auto rf = bootstrap.Fuse->SendRequest(rq);
            bootstrap.GetNodeAttrRequestEvent.WaitT(WaitTimeout);
            const ui64 oldSize = bootstrap.NodeSize;
            bootstrap.SetSize(12_KB);

            NProto::TGetNodeAttrResponse r;
            auto& n = *r.MutableNode();
            n.SetId(bootstrap.NodeId);
            n.SetSize(oldSize);
            bootstrap.GetNodeAttrResponse.SetValue(std::move(r));
            rf.GetValue(WaitTimeout);

            //
            // SetAttr completed while we were processing GetNodeAttr - the
            // result is stale, shouldn't be cached (attr_timeout should be 0).
            //

            UNIT_ASSERT_VALUES_EQUAL(oldSize, rq->Out->Body.attr.size);
            UNIT_ASSERT_VALUES_EQUAL(0, rq->Out->Body.attr_valid);
            UNIT_ASSERT_VALUES_EQUAL(0, rq->Out->Body.attr_valid_nsec);
        }
    }

    Y_UNIT_TEST(
        ShouldNotAllowGuestAttrCachingUponListNodesAndAttrModificationRace)
    {
        TAttrRaceBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        const TString name = "file";

        auto doTest = [&] (const std::function<void()>& w) {
            auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(
                bootstrap.ParentNodeId);
            UNIT_ASSERT(handle.Wait(WaitTimeout));
            auto handleId = handle.GetValue();

            auto rq = std::make_shared<TReadDirRequest>(
                bootstrap.ParentNodeId,
                handleId);
            auto rf = bootstrap.Fuse->SendRequest(rq);
            bootstrap.ListNodesRequestEvent.WaitT(WaitTimeout);
            const ui64 oldSize = bootstrap.NodeSize;
            w();

            NProto::TListNodesResponse r;
            r.AddNames(name);
            auto* node = r.AddNodes();
            node->SetId(bootstrap.NodeId);
            node->SetType(NProto::E_REGULAR_NODE);
            node->SetSize(oldSize);
            bootstrap.ListNodesResponse.SetValue(std::move(r));
            rf.GetValue(WaitTimeout);

            //
            // Write completed while we were processing ListNodes - the result
            // is stale, shouldn't be cached (attr_timeout should be 0).
            //

            auto buf = TStringBuf(
                reinterpret_cast<const char*>(rq->Out->Data()),
                rq->Out->Header.len - sizeof(rq->Out->Header));

            // Skip "." and ".." entries
            UNIT_ASSERT_LE(
                sizeof(fuse_direntplus) + dotSize + dotDotSize,
                buf.size());
            buf = buf.Skip(dotSize + dotDotSize);

            const auto* de =
                reinterpret_cast<const fuse_direntplus*>(buf.data());

            UNIT_ASSERT_VALUES_EQUAL(oldSize, de->entry_out.attr.size);
            UNIT_ASSERT_VALUES_EQUAL(0, de->entry_out.attr_valid);
            UNIT_ASSERT_VALUES_EQUAL(0, de->entry_out.attr_valid_nsec);

            auto close = bootstrap.Fuse->SendRequest<TReleaseDirRequest>(
                bootstrap.ParentNodeId,
                handleId);
            UNIT_ASSERT_NO_EXCEPTION(close.GetValue(WaitTimeout));
        };

        doTest([&] () {
            bootstrap.Write(bootstrap.NodeSize, 4_KB);
        });

        doTest([&] () {
            bootstrap.Allocate(bootstrap.NodeSize, 4_KB);
        });

        doTest([&] () {
            bootstrap.SetSize(12_KB);
        });
    }

    Y_UNIT_TEST(ShouldListWithDirectoryContentChunkAtNameBoundary)
    {
        TBootstrap bootstrap;
        bootstrap.Service->ListNodesHandler = [&](auto, auto)
        {
            NProto::TListNodesResponse r;
            for (ui32 ino = 2; ino < 100; ++ino) {
                r.AddNames(TStringBuilder() << "f" << ino);
                auto* node = r.AddNodes();
                node->SetId(ino);
                node->SetType(NProto::E_REGULAR_NODE);
                node->SetSize(4_KB);
            }
            return MakeFuture(std::move(r));
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        const ui64 parentNodeId = 1;

        auto handle =
            bootstrap.Fuse->SendRequest<TOpenDirRequest>(parentNodeId);
        UNIT_ASSERT(handle.Wait(WaitTimeout));
        auto handleId = handle.GetValue();

        const ui64 smallDirentSize =
            sizeof(fuse_direntplus) + sizeof(ui64);
        const ui64 direntWithoutNameSize = sizeof(fuse_direntplus);
        const ui64 size = dotSize + dotDotSize
            + 10 * smallDirentSize + direntWithoutNameSize;
        auto rq = std::make_shared<TReadDirRequest>(
            parentNodeId,
            handleId,
            0 /* offset */,
            size);
        auto rf = bootstrap.Fuse->SendRequest(rq);
        rf.GetValue(WaitTimeout);

        auto buf = TStringBuf(
            reinterpret_cast<const char*>(rq->Out->Data()),
            rq->Out->Header.len - sizeof(rq->Out->Header));

        // Skip "." and ".." entries
        UNIT_ASSERT_LE(
            sizeof(fuse_direntplus) + dotSize + dotDotSize,
            buf.size());
        buf = buf.Skip(dotSize + dotDotSize);

        for (ui32 ino = 2; ino < 12; ++ino) {
            UNIT_ASSERT_GE(buf.Size(), sizeof(fuse_direntplus));

            const auto* de =
                reinterpret_cast<const fuse_direntplus*>(buf.data());

            UNIT_ASSERT_VALUES_EQUAL(4_KB, de->entry_out.attr.size);
            UNIT_ASSERT_VALUES_EQUAL(
                TStringBuilder() << "f" << ino,
                TStringBuf(de->dirent.name, de->dirent.namelen));

            buf = buf.Skip(AlignUp<ui64>(
                sizeof(fuse_direntplus) + de->dirent.namelen,
                sizeof(uint64_t)));
        }

        //
        // Checking that our test did what's expected - generated a dirent
        // without a name.
        //

        UNIT_ASSERT_VALUES_EQUAL(direntWithoutNameSize, buf.Size());

        //
        // This incomplete dirent shouldn't cause any errors.
        //

        auto counters = bootstrap.Counters
            ->FindSubgroup("component", "fs_ut")
            ->FindSubgroup("request", "ListNodes");
        // counters->OutputPlainText(Cerr);
        UNIT_ASSERT_VALUES_EQUAL(
            0,
            counters->GetCounter("Errors")->GetAtomic());
        UNIT_ASSERT_VALUES_EQUAL(
            0,
            counters->GetCounter("Errors/Fatal")->GetAtomic());

        auto close = bootstrap.Fuse->SendRequest<TReleaseDirRequest>(
            parentNodeId,
            handleId);
        UNIT_ASSERT_NO_EXCEPTION(close.GetValue(WaitTimeout));
    }

    Y_UNIT_TEST(ShouldHandleWriteBackCacheFlushFailures)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        const ui64 NodeId = 123;

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        // Any attempt to flush data will fail
        bootstrap.Service->WriteDataHandler = [&](auto, const auto& rq)
        {
            NProto::TWriteDataResponse result;
            if (rq->GetOffset() == 0) {
                result = TErrorResponse(E_FAIL, "xxx");
            }
            return MakeFuture(result);
        };

        bootstrap.Service->DestroyHandleHandler = [&](auto, const auto&)
        {
            NProto::TDestroyHandleResponse result;
            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(
            NodeId, NodeId + 1, 0, CreateBuffer(4096, 'a'));
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));

        // 1. SetAttr
        auto setAttr = std::make_shared<TSetAttrRequest>(NodeId, NodeId + 1);
        setAttr->In->Body.valid = FUSE_SET_ATTR_SIZE;
        setAttr->In->Body.size = 11;
        UNIT_ASSERT(bootstrap.Fuse->SendRequest(setAttr).Wait(WaitTimeout));
        UNIT_ASSERT(setAttr->Out->Header.error);

        // 2. Flush
        auto flush = std::make_shared<TFlushRequest>(NodeId, NodeId + 1);
        UNIT_ASSERT(bootstrap.Fuse->SendRequest(flush).Wait(WaitTimeout));
        UNIT_ASSERT(flush->Out->Header.error);

        // 3. Write with DSYNC - should fail because flush fails
        auto writeDSync = std::make_shared<TWriteRequest>(
            NodeId,
            NodeId + 1,
            11,
            CreateBuffer(4096, 'a'));
        writeDSync->In->Body.flags |= O_DSYNC;
        writeDSync->In->Body.flags |= O_WRONLY;
        UNIT_ASSERT(bootstrap.Fuse->SendRequest(writeDSync).Wait(WaitTimeout));
        UNIT_ASSERT(writeDSync->Out->Header.error);

        // 4. Release - should fail because data is dropped
        auto release = std::make_shared<TReleaseRequest>(
            NodeId,
            NodeId + 1,
            O_WRONLY);
        UNIT_ASSERT(bootstrap.Fuse->SendRequest(release).Wait(WaitTimeout));
        UNIT_ASSERT(release->Out->Header.error);

        // 5. Release again - success because of no data in the cache
        auto release2 = std::make_shared<TReleaseRequest>(
            NodeId,
            NodeId + 1,
            O_WRONLY);
        UNIT_ASSERT(bootstrap.Fuse->SendRequest(release2).Wait(WaitTimeout));
        UNIT_ASSERT(!release2->Out->Header.error);
    }

    Y_UNIT_TEST(ShouldUseWriteBackCacheBarriers)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        const ui64 NodeId = 123;
        const ui64 HandleId = 456;

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        std::atomic<int> writeDataCalled = 0;

        bootstrap.Service->WriteDataHandler = [&](auto, const auto&)
        {
            writeDataCalled++;
            NProto::TWriteDataResponse result;
            return MakeFuture(result);
        };

        auto setNodeAttrCalled = NewPromise();
        auto setNodeAttrPromise = NewPromise<NProto::TSetNodeAttrResponse>();

        bootstrap.Service->SetNodeAttrHandler = [&](auto, const auto&)
        {
            setNodeAttrCalled.SetValue();
            return setNodeAttrPromise.GetFuture();
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        // Initiate a request that acquires a barrier
        {
            auto rq = std::make_shared<TSetAttrRequest>(NodeId, HandleId);
            rq->In->Body.valid = FUSE_SET_ATTR_SIZE;
            rq->In->Body.size = 42;
            bootstrap.Fuse->SendRequest(rq);
            setNodeAttrCalled.GetFuture().GetValue(WaitTimeout);
        }

        // Add some data to write-back cache
        {
            auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(
                NodeId,
                NodeId + 1,
                0,
                "abc");
            UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
            UNIT_ASSERT_VALUES_EQUAL(0, writeDataCalled.load());
        }

        auto flush =
            bootstrap.Fuse->SendRequest<TFlushRequest>(NodeId, NodeId + 1);

        // WriteBackCache will not execute flush until all barriers are released
        // There is no reliable way to ensure that flush request was processed
        // by write-back cache
        // TODO(#1751): add a metric for this
        Sleep(TDuration::Seconds(1));

        UNIT_ASSERT(!flush.HasValue());
        UNIT_ASSERT_VALUES_EQUAL(0, writeDataCalled.load());

        NProto::TSetNodeAttrResponse response;
        response.MutableNode()->SetId(NodeId);
        response.MutableNode()->SetSize(42);
        setNodeAttrPromise.SetValue(std::move(response));

        UNIT_ASSERT_NO_EXCEPTION(flush.GetValue(WaitTimeout));
        UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());
    }

    Y_UNIT_TEST(DirectReadAndWritesShouldTriggerWriteBackCacheFlush)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        const ui64 NodeId = 123;
        const ui64 HandleId = 456;

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        std::atomic<int> readDataCalled = 0;
        std::atomic<int> writeDataCalled = 0;

        bootstrap.Service->ReadDataHandler = [&](auto, const auto&)
        {
            readDataCalled++;
            NProto::TReadDataResponse result;
            return MakeFuture(result);
        };

        bootstrap.Service->WriteDataHandler = [&](auto, const auto&)
        {
            writeDataCalled++;
            NProto::TWriteDataResponse result;
            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto addSomeData = [&]()
        {
            auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(
                NodeId,
                HandleId,
                1000,
                "abc");
            UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        };

        addSomeData();

        {
            // Cached read
            auto rq = std::make_shared<TReadRequest>(NodeId, HandleId, 0, 11);
            auto read = bootstrap.Fuse->SendRequest(rq);
            UNIT_ASSERT_NO_EXCEPTION(read.GetValue(WaitTimeout));
            UNIT_ASSERT_VALUES_EQUAL(1, readDataCalled.load());
            UNIT_ASSERT_VALUES_EQUAL(0, writeDataCalled.load());
        }

        {
            // Direct read
            auto rq = std::make_shared<TReadRequest>(NodeId, HandleId, 0, 11);
            rq->In->Body.flags |= O_DIRECT;
            auto read = bootstrap.Fuse->SendRequest(rq);
            UNIT_ASSERT_NO_EXCEPTION(read.GetValue(WaitTimeout));
            UNIT_ASSERT_VALUES_EQUAL(2, readDataCalled.load());
            UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());
        }

        {
            // Cached write
            auto rq = std::make_shared<TWriteRequest>(NodeId, HandleId, 0, "a");
            auto write = bootstrap.Fuse->SendRequest(rq);
            UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
            UNIT_ASSERT_VALUES_EQUAL(1, writeDataCalled.load());
        }

        {
            // Direct write
            auto rq = std::make_shared<TWriteRequest>(NodeId, HandleId, 0, "a");
            rq->In->Body.flags |= O_DIRECT;
            auto write = bootstrap.Fuse->SendRequest(rq);
            UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
            UNIT_ASSERT_VALUES_EQUAL(3, writeDataCalled.load());
        }

        addSomeData();

        {
            // O_SYNC write
            auto rq = std::make_shared<TWriteRequest>(NodeId, HandleId, 0, "a");
            rq->In->Body.flags |= O_SYNC;
            auto write = bootstrap.Fuse->SendRequest(rq);
            UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
            UNIT_ASSERT_VALUES_EQUAL(5, writeDataCalled.load());
        }

        addSomeData();

        {
            // O_DSYNC write
            auto rq = std::make_shared<TWriteRequest>(NodeId, HandleId, 0, "a");
            rq->In->Body.flags |= O_DSYNC;
            auto write = bootstrap.Fuse->SendRequest(rq);
            UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
            UNIT_ASSERT_VALUES_EQUAL(7, writeDataCalled.load());
        }
    }

    Y_UNIT_TEST(ShouldNotReportEnormousMaxTimeWhenCancellingRequest)
    {
        NProto::TFileStoreFeatures features;

        // StopAsync and GetAttrRequest are executed concurrently:
        // - if StopAsync is completed before GetAttrRequest, it will not be
        //   processed at all;
        // - if GetAttrRequest is completed before StopAsync, it will not be
        //   canceled and will return with success.
        //
        // We use WriteBackCache in order to prevent race and enforce strict
        // ordering between StopAsync and GetAttrRequest execution.
        //
        // WriteBackCache::FlushAll is executed after CompletionQueue is stopped
        // (new requests will be canceled) and before FUSE loop is unmounted.
        //
        // We catch the moment when FlushAll is called by listening to
        // WriteData requests.

        features.SetServerWriteBackCacheEnabled(true);

        const ui64 NodeId = 123;
        const ui64 HandleId = 456;

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        auto writeDataCalledPromise = NewPromise();
        auto responsePromise = NewPromise<NProto::TWriteDataResponse>();

        bootstrap.Service->WriteDataHandler = [&](auto, const auto&)
        {
            writeDataCalledPromise.SetValue();
            return responsePromise.GetFuture();
        };

        bootstrap.Start();

        {
            // Cached write
            auto rq = std::make_shared<TWriteRequest>(NodeId, HandleId, 0, "a");
            auto write = bootstrap.Fuse->SendRequest(rq);
            UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));
        }

        auto stopFuture = bootstrap.StopAsync();

        // StopAsync triggers FlushAll
        writeDataCalledPromise.GetFuture().GetValue(WaitTimeout);

        auto future = bootstrap.Fuse->SendRequest<TGetAttrRequest>(NodeId + 1);
        future.Wait(WaitTimeout);

        // Request is cancelled
        UNIT_ASSERT_EXCEPTION(future.GetValue(WaitTimeout), yexception);

        bootstrap.StatsRegistry->UpdateStats(false);

        auto counters = bootstrap.Counters
            ->FindSubgroup("component", TString{MetricsComponent} + "_fs")
            ->FindSubgroup("host", "cluster")
            ->FindSubgroup("filesystem", FileSystemId)
            ->FindSubgroup("client", "")
            ->FindSubgroup("cloud", "")
            ->FindSubgroup("folder", "")
            ->FindSubgroup("request", "GetNodeAttr");

        UNIT_ASSERT_VALUES_EQUAL(1, counters->GetCounter("Errors")->GetAtomic());

        const auto maxTime = counters->GetCounter("MaxTime")->GetAtomic();
        UNIT_ASSERT_LE_C(
            maxTime,
            1000000000000,
            "got maxTime " << maxTime);

        // Complete FlushAll so StopAsync can finish
        responsePromise.SetValue(NProto::TWriteDataResponse());

        UNIT_ASSERT(stopFuture.Wait(WaitTimeout));
    }

    Y_UNIT_TEST(WriteBackCacheShouldReportMetrics)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        const ui64 NodeId = 123;
        const ui64 HandleId = 456;

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features);

        bootstrap.Service->ReadDataHandler = [&](auto, const auto&)
        {
            NProto::TReadDataResponse result;
            return MakeFuture(result);
        };

        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto read =
            bootstrap.Fuse->SendRequest<TReadRequest>(NodeId, HandleId, 0, 11);
        UNIT_ASSERT_NO_EXCEPTION(read.GetValue(WaitTimeout));


        auto fsCounters = bootstrap.Counters
            ->FindSubgroup("component", "fs_ut_fs")
            ->FindSubgroup("host", "cluster")
            ->FindSubgroup("filesystem", FileSystemId)
            ->FindSubgroup("client", "")
            ->FindSubgroup("cloud", "")
            ->FindSubgroup("folder", "")
            ->FindSubgroup("module", "WriteBackCache");

        auto totalCounters = bootstrap.Counters
            ->FindSubgroup("component", "fs_ut")
            ->FindSubgroup("module", "WriteBackCache");

        bootstrap.ModuleStatsRegistry->UpdateStats(true);

        UNIT_ASSERT_VALUES_EQUAL(
            1,
            fsCounters->GetCounter("ReadData_CacheMiss")->GetAtomic());

        UNIT_ASSERT_VALUES_EQUAL(
            1,
            totalCounters->GetCounter("ReadData_CacheMiss")->GetAtomic());

        bootstrap.Stop();

        UNIT_ASSERT(!bootstrap.Counters->FindSubgroup("component", "fs_ut_fs")
                         ->FindSubgroup("host", "cluster")
                         ->FindSubgroup("filesystem", FileSystemId)
                         ->FindSubgroup("client", ""));

        UNIT_ASSERT(!bootstrap.Counters
            ->FindSubgroup("component", "fs_ut")
            ->FindSubgroup("module", "WriteBackCache"));
    }

    Y_UNIT_TEST(ShouldNotDeadlockWhenFlushCompletionFiredInScheduler)
    {
        // Deadlock previously happened when:
        // - automatic flush was executed on a scheduler thread;
        // - at the same time, Stop was requested and triggered FlushAllData;
        // - FlushAllData continuation was called on the same scheduler thread;
        // - Stop continuation was called on the same scheduler thread and tried
        //   to stop scheduler.

        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        ui32 automaticFlushPeriodMs = 1;
        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features,
            0,
            automaticFlushPeriodMs);

        const ui64 nodeId = 123;
        const ui64 handleId = 456;

        auto writeDataCalled = NewPromise();
        auto writeDataPromise = NewPromise<NProto::TWriteDataResponse>();

        bootstrap.Service->WriteDataHandler = [&](auto callContext, auto)
        {
            UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId);
            writeDataCalled.SetValue();

            // Returns a pending future that will be completed manually on a
            // scheduler thread after verifying FlushAll is in progress
            return writeDataPromise.GetFuture();
        };

        bootstrap.Start();
        Y_DEFER
        {
            bootstrap.Stop();
        };

        // write request without O_DIRECT should go to write cache
        auto reqWrite = std::make_shared<TWriteRequest>(
            nodeId,
            handleId,
            0,
            CreateBuffer(4096, 'a'));
        reqWrite->In->Body.flags |= O_WRONLY;
        auto write = bootstrap.Fuse->SendRequest<TWriteRequest>(reqWrite);
        UNIT_ASSERT_NO_EXCEPTION(write.GetValue(WaitTimeout));

        // Wait until automatic flush is called
        UNIT_ASSERT_NO_EXCEPTION(
            writeDataCalled.GetFuture().GetValue(WaitTimeout));

        auto counter = bootstrap.Counters->FindSubgroup("component", "fs_ut_fs")
                           ->FindSubgroup("host", "cluster")
                           ->FindSubgroup("filesystem", FileSystemId)
                           ->FindSubgroup("client", "")
                           ->FindSubgroup("cloud", "")
                           ->FindSubgroup("folder", "")
                           ->FindSubgroup("module", "WriteBackCache")
                           ->GetCounter("FlushAllRequests_InProgressCount");

        // Automatic flush does not create a request internally
        UNIT_ASSERT_VALUES_EQUAL(0, counter->GetAtomic());

        auto stopFuture = bootstrap.StopAsync();

        // Wait until FlushAll is triggered in session destroy handler
        UNIT_ASSERT(WaitForCondition(
            WaitTimeout,
            [&]
            {
                // Stats are updated automatically with 1-second interval
                // Update it manually to speed up the test
                bootstrap.ModuleStatsRegistry->UpdateStats(true);
                return counter->GetAtomic() > 0;
            }));

        // WriteData completion should be done in a Scheduler thread, otherwise
        // the deadlock won't reproduce
        bootstrap.Scheduler->Schedule(
            TInstant::Zero(),
            [writeDataPromise]() mutable { writeDataPromise.SetValue({}); });

        UNIT_ASSERT(stopFuture.Wait(WaitTimeout));
    }

    Y_UNIT_TEST(ShouldHandleServerWriteBackCacheFlushWritesInParallelEnabledFeature)
    {
        // Sends 3 WriteData requests to WriteBackCache: [0, 3), [6, 9), [3, 6)
        //
        // WriteBackCache uses greedy algorithm to build WriteData requests
        // for flush and takes requests from the unflushed queue while the
        // condition is met.
        //
        // When parallel writes are disabled:
        // - after taking [0, 3) and [6, 9) there will be 2 WriteData requests
        //   so it will flush only [0, 3)
        // - then it takes [6, 9) and [3, 6) then are consolidated to [3, 9)
        //
        // When parallel writes are enabled:
        // - take [0, 3) and [6, 9) - 2 WriteData requests in parallel are ok
        // - take [6, 9) - consolidate to [0, 9)

        auto test = [&](bool flushWritesInParallelEnabled)
        {
            NProto::TFileStoreFeatures features;
            features.SetServerWriteBackCacheEnabled(true);
            features.SetServerWriteBackCacheFlushWritesInParallelEnabled(
                flushWritesInParallelEnabled);

            TBootstrap bootstrap(
                CreateWallClockTimer(),
                CreateScheduler(),
                 features,
                 /* handleOpsQueueSize = */ 0,
                 /* writeBackCacheAutomaticFlushPeriodMs = */ 0);

            const ui64 nodeId = 123;
            const ui64 handleId = 456;

            std::atomic<ui64> writeDataCalled = 0;

            bootstrap.Service->WriteDataHandler = [&](auto, auto)
            {
                writeDataCalled++;
                return MakeFuture<NProto::TWriteDataResponse>({});
            };

            bootstrap.Start();
            Y_DEFER
            {
                bootstrap.Stop();
            };

            auto write1 = bootstrap.Fuse->SendRequest<TWriteRequest>(
                nodeId,
                handleId,
                0,
                "abc");

            UNIT_ASSERT_NO_EXCEPTION(write1.GetValue(WaitTimeout));

            auto write2 = bootstrap.Fuse->SendRequest<TWriteRequest>(
                nodeId,
                handleId,
                6,
                "ghi");

            UNIT_ASSERT_NO_EXCEPTION(write2.GetValue(WaitTimeout));

            auto write3 = bootstrap.Fuse->SendRequest<TWriteRequest>(
                nodeId,
                handleId,
                3,
                "def");

            UNIT_ASSERT_NO_EXCEPTION(write3.GetValue(WaitTimeout));

            auto flush =
                bootstrap.Fuse->SendRequest<TFlushRequest>(nodeId, handleId);
            UNIT_ASSERT_NO_EXCEPTION(flush.GetValue(WaitTimeout));

            return writeDataCalled.load();
        };

        UNIT_ASSERT_VALUES_EQUAL(1, test(true));
        UNIT_ASSERT_VALUES_EQUAL(2, test(false));
    }

    void TestListNodesAndEntryInvalidationRace(
        const TString& listedName,
        NProto::ENodeType nodeType,
        const std::function<void(TAttrRaceBootstrap&)>& mutate)
    {
        TAttrRaceBootstrap bootstrap;
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto getFirstListedDirent = [] (const TReadDirRequest& request)
        {
            auto buf = TStringBuf(
                reinterpret_cast<const char*>(request.Out->Data()),
                request.Out->Header.len - sizeof(request.Out->Header));

            // Skip "." and ".." entries
            UNIT_ASSERT_LE(
                sizeof(fuse_direntplus) + dotSize + dotDotSize,
                buf.size());
            buf = buf.Skip(dotSize + dotDotSize);

            return reinterpret_cast<const fuse_direntplus*>(buf.data());
        };

        auto handle = bootstrap.Fuse->SendRequest<TOpenDirRequest>(
            bootstrap.ParentNodeId);
        UNIT_ASSERT(handle.Wait(WaitTimeout));
        const ui64 handleId = handle.GetValue();

        auto rq = std::make_shared<TReadDirRequest>(
            bootstrap.ParentNodeId,
            handleId);
        auto rf = bootstrap.Fuse->SendRequest(rq);
        UNIT_ASSERT(bootstrap.ListNodesRequestEvent.WaitT(WaitTimeout));

        mutate(bootstrap);

        NProto::TListNodesResponse response;
        response.AddNames(listedName);
        auto* node = response.AddNodes();
        node->SetId(bootstrap.NodeId);
        node->SetType(nodeType);
        node->SetSize(bootstrap.NodeSize);
        bootstrap.ListNodesResponse.SetValue(std::move(response));
        rf.GetValue(WaitTimeout);

        const auto* de = getFirstListedDirent(*rq);

        UNIT_ASSERT_VALUES_EQUAL(bootstrap.NodeId, de->entry_out.nodeid);
        UNIT_ASSERT_GT(de->entry_out.attr_valid, 0);
        UNIT_ASSERT_VALUES_EQUAL(0, de->entry_out.entry_valid);
        UNIT_ASSERT_VALUES_EQUAL(0, de->entry_out.entry_valid_nsec);

        auto close = bootstrap.Fuse->SendRequest<TReleaseDirRequest>(
            bootstrap.ParentNodeId,
            handleId);
        UNIT_ASSERT_NO_EXCEPTION(close.GetValue(WaitTimeout));
    }

    Y_UNIT_TEST(ShouldNotAllowGuestEntryCachingUponListNodesAndUnlinkRace)
    {
        TestListNodesAndEntryInvalidationRace(
            "file",
            NProto::E_REGULAR_NODE,
            [] (TAttrRaceBootstrap& bootstrap) {
                bootstrap.Unlink("file");
            });
    }

    Y_UNIT_TEST(ShouldNotAllowGuestEntryCachingUponListNodesAndRmDirRace)
    {
        TestListNodesAndEntryInvalidationRace(
            "dir",
            NProto::E_DIRECTORY_NODE,
            [] (TAttrRaceBootstrap& bootstrap) {
                bootstrap.RmDir("dir");
            });
    }

    Y_UNIT_TEST(
        ShouldNotAllowGuestEntryCachingUponListNodesAndRenameSourceRace)
    {
        TestListNodesAndEntryInvalidationRace(
            "file",
            NProto::E_REGULAR_NODE,
            [] (TAttrRaceBootstrap& bootstrap) {
                bootstrap.Rename("file", "renamed");
            });
    }

    Y_UNIT_TEST(
        ShouldNotAllowGuestEntryCachingUponListNodesAndRenameTargetRace)
    {
        TestListNodesAndEntryInvalidationRace(
            "target",
            NProto::E_REGULAR_NODE,
            [] (TAttrRaceBootstrap& bootstrap) {
                bootstrap.Rename("file", "target");
            });
    }

    Y_UNIT_TEST(ShouldNotCacheRequestsOnWriteBackCacheFlushBatchBackpressure)
    {
        NProto::TFileStoreFeatures features;
        features.SetServerWriteBackCacheEnabled(true);

        // Disable parallel writes in order to ensure that each flush batch
        // will consist of a single WriteData request
        features.SetServerWriteBackCacheFlushWritesInParallelEnabled(false);

        auto firstWriteDataPromise = NewPromise<NProto::TWriteDataResponse>();
        auto secondWriteDataPromise = NewPromise<NProto::TWriteDataResponse>();
        auto restWriteDataPromise = NewPromise<NProto::TWriteDataResponse>();
        std::atomic<int> writeDataCalled = 0;

        TBootstrap bootstrap(
            CreateWallClockTimer(),
            CreateScheduler(),
            features,
            /* handleOpsQueueSize = */ 0,
            /* writeBackCacheAutomaticFlushPeriodMs = */ 0);

        bootstrap.Start();
        Y_DEFER
        {
            bootstrap.Stop();
        };

        bootstrap.Service->WriteDataHandler = [&](auto, auto)
        {
            switch (writeDataCalled++) {
                case 0:
                    return firstWriteDataPromise.GetFuture();
                case 1:
                    return secondWriteDataPromise.GetFuture();
                default:
                    return restWriteDataPromise.GetFuture();
            }
        };

        // WriteBackCacheMaxQueuedFlushBatchesPerNode is non-zero by
        // default - eventually backpressure will occur
        constexpr ui32 maxRequestCount = 1500;

        auto counterGroup =
            bootstrap.Counters->FindSubgroup("component", "fs_ut_fs")
                ->FindSubgroup("host", "cluster")
                ->FindSubgroup("filesystem", FileSystemId)
                ->FindSubgroup("client", "")
                ->FindSubgroup("cloud", "")
                ->FindSubgroup("folder", "")
                ->FindSubgroup("module", "WriteBackCache");

        auto pendingQueueCount = counterGroup->GetCounter("PendingQueue_Count");

        auto unflushedQueueCount =
            counterGroup->GetCounter("UnflushedQueue_Count");

        auto backpressureCount =
            counterGroup->GetCounter("NodesWithBackpressure_Count");

        TFuture<ui32> writeFuture;

        for (ui32 i = 1; i <= maxRequestCount; i++) {
            // Requests should not overlap nor touch in order to be put into
            // separate flush batches
            writeFuture = bootstrap.Fuse->SendRequest<TWriteRequest>(
                1,
                101,
                i * 100,
                "abc");

            // Request should be put in either unflushed or pending queue
            UNIT_ASSERT(WaitForCondition(
                WaitTimeout,
                [&]()
                {
                    bootstrap.ModuleStatsRegistry->UpdateStats(true);
                    return unflushedQueueCount->Val() == i ||
                           pendingQueueCount->Val() == 1;
                }));

            if (pendingQueueCount->Val() == 1) {
                // This happens when backpressure is applied
                break;
            } else {
                UNIT_ASSERT(writeFuture.Wait(WaitTimeout));
            }
        }

        UNIT_ASSERT_VALUES_EQUAL(1, pendingQueueCount->Val());
        UNIT_ASSERT_VALUES_EQUAL(1, backpressureCount->Val());
        UNIT_ASSERT(!writeFuture.HasValue());

        firstWriteDataPromise.SetValue({});

        UNIT_ASSERT(writeFuture.Wait(WaitTimeout));

        bootstrap.ModuleStatsRegistry->UpdateStats(true);
        UNIT_ASSERT_VALUES_EQUAL(0, pendingQueueCount->Val());
        UNIT_ASSERT_VALUES_EQUAL(1, backpressureCount->Val());

        secondWriteDataPromise.SetValue({});

        UNIT_ASSERT(WaitForCondition(
            WaitTimeout,
            [&]()
            {
                bootstrap.ModuleStatsRegistry->UpdateStats(true);
                return backpressureCount->Val() == 0;
            }));

        restWriteDataPromise.SetValue({});
    }

    NProto::TFileStoreFeatures AvailabilityTrackingFeatures()
    {
        NProto::TFileStoreFeatures features;
        features.SetAvailabilityTrackingEnabled(true);
        features.SetAvailabilityTrackingInterval(15000);   // in ms
        return features;
    }

    Y_UNIT_TEST(ShouldTrackAvailabilityOfSuccessfulRequests)
    {
        auto timer = std::make_shared<TTestTimer>();
        TBootstrap bootstrap{
            timer,
            CreateScheduler(),
            AvailabilityTrackingFeatures()};
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        bootstrap.Service->CreateHandleHandler =
            [] (auto callContext, auto request)
        {
            Y_UNUSED(callContext);
            Y_UNUSED(request);
            NProto::TCreateHandleResponse result;
            result.SetHandle(1);
            result.MutableNodeAttr()->SetId(2);
            result.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
            return MakeFuture(result);
        };

        // start the first interval
        timer->AdvanceTime(TDuration::MilliSeconds(15000));
        bootstrap.StatsRegistry->UpdateStats(false /* updatePercentiles */);

        // The completed wait below guarantees the availability events were
        // recorded before the clock advances, so the success is classified
        // in the first interval - the assertions cover the request itself,
        // not an idle interval.
        auto handle = bootstrap.Fuse->SendRequest<TCreateHandleRequest>(
            "/file1",
            RootNodeId);
        UNIT_ASSERT(handle.Wait(WaitTimeout));

        // roll the first interval
        timer->AdvanceTime(TDuration::MilliSeconds(15000));
        bootstrap.StatsRegistry->UpdateStats(false /* updatePercentiles */);

        auto counters = bootstrap.GetFileSystemStatsCounters();
        UNIT_ASSERT(counters);
        UNIT_ASSERT_VALUES_EQUAL(
            1,
            counters->GetCounter(
                "Availability_TotalIntervals",
                true)->Val());
        UNIT_ASSERT_VALUES_EQUAL(
            1,
            counters->GetCounter(
                "Availability_AvailableIntervals",
                true)->Val());

        // the fuse create callback is published under request=create
        auto createCounters = counters->FindSubgroup("request", "create");
        UNIT_ASSERT(createCounters);
        UNIT_ASSERT_VALUES_EQUAL(
            1,
            createCounters->GetCounter(
                "Availability_AvailableIntervals",
                true)->Val());
    }

    Y_UNIT_TEST(ShouldTrackAvailabilityOfEioRequests)
    {
        auto timer = std::make_shared<TTestTimer>();
        TBootstrap bootstrap{
            timer,
            CreateScheduler(),
            AvailabilityTrackingFeatures()};
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        bootstrap.Service->CreateHandleHandler =
            [] (auto callContext, auto request)
        {
            Y_UNUSED(callContext);
            Y_UNUSED(request);
            // E_IO is delivered to the guest as errno EIO
            return MakeFuture(
                NProto::TCreateHandleResponse(TErrorResponse(E_IO)));
        };

        // start the first interval
        timer->AdvanceTime(TDuration::MilliSeconds(15000));
        bootstrap.StatsRegistry->UpdateStats(false /* updatePercentiles */);

        // roll the first interval and start the second interval
        timer->AdvanceTime(TDuration::MilliSeconds(15000));
        bootstrap.StatsRegistry->UpdateStats(false /* updatePercentiles */);

        // request is started in the second interval
        auto handle = bootstrap.Fuse->SendRequest<TCreateHandleRequest>(
            "/file1",
            RootNodeId);
        UNIT_ASSERT(handle.Wait(WaitTimeout));

        // roll the second interval
        //
        // request returned Eio => interval unvailable
        timer->AdvanceTime(TDuration::MilliSeconds(15000));
        bootstrap.StatsRegistry->UpdateStats(false /* updatePercentiles */);

        auto counters = bootstrap.GetFileSystemStatsCounters();
        UNIT_ASSERT(counters);
        UNIT_ASSERT_VALUES_EQUAL(
            2,
            counters->GetCounter(
                "Availability_TotalIntervals",
                true)->Val());
        UNIT_ASSERT_VALUES_EQUAL(
            1,
            counters->GetCounter(
                "Availability_UnavailableIntervals",
                true)->Val());
        UNIT_ASSERT_VALUES_EQUAL(
            0,
            counters->GetCounter(
                "Availability_LastIntervalAvailable")->Val());

        auto createCounters = counters->FindSubgroup("request", "create");
        UNIT_ASSERT(createCounters);
        UNIT_ASSERT_VALUES_EQUAL(
            1,
            createCounters->GetCounter(
                "Availability_UnavailableIntervals",
                true)->Val());
    }

    Y_UNIT_TEST(ShouldTrackAvailabilityOfHungRequests)
    {
        auto timer = std::make_shared<TTestTimer>();
        TBootstrap bootstrap{
            timer,
            CreateScheduler(),
            AvailabilityTrackingFeatures()};
        bootstrap.Start();
        Y_DEFER {
            bootstrap.Stop();
        };

        auto hungPromise = NewPromise<NProto::TCreateHandleResponse>();
        auto requestReceived = NewPromise<void>();
        bootstrap.Service->CreateHandleHandler =
            [&] (auto callContext, auto request)
        {
            Y_UNUSED(callContext);
            Y_UNUSED(request);
            requestReceived.TrySetValue();
            return hungPromise.GetFuture();
        };

        // start the first interval
        timer->AdvanceTime(TDuration::MilliSeconds(15000));
        bootstrap.StatsRegistry->UpdateStats(false /* updatePercentiles */);

        // request is started in the first interval
        auto handle = bootstrap.Fuse->SendRequest<TCreateHandleRequest>(
            "/file1",
            RootNodeId);

        // The request is processed on the fuse loop thread asynchronously:
        // wait until it has actually reached the service - its availability
        // registration happens earlier on the same path - before advancing
        // the fake clock. Otherwise the registration races the advances and
        // may see a later fake time, shifting the request into a later
        // interval.
        UNIT_ASSERT(requestReceived.GetFuture().Wait(WaitTimeout));

        // it is fresh-pending (neutral) in the first interval
        //
        // roll the first interval and start the second interval
        timer->AdvanceTime(TDuration::MilliSeconds(15000));
        bootstrap.StatsRegistry->UpdateStats(false /* updatePercentiles */);

        // request hung through the second interval => unavailable
        timer->AdvanceTime(TDuration::MilliSeconds(15000));
        bootstrap.StatsRegistry->UpdateStats(false /* updatePercentiles */);

        // Unblock the request before the assertions so that a failing
        // assertion cannot unwind into Stop() with the request still
        // outstanding.
        NProto::TCreateHandleResponse result;
        result.SetHandle(1);
        result.MutableNodeAttr()->SetId(2);
        result.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE);
        hungPromise.SetValue(result);
        UNIT_ASSERT(handle.Wait(WaitTimeout));

        // Two intervals are classified: the first interval saw the request
        // fresh-pending (neutral, no failure evidence) => available, the
        // second interval saw it hung throughout => unavailable.
        auto counters = bootstrap.GetFileSystemStatsCounters();
        UNIT_ASSERT(counters);
        UNIT_ASSERT_VALUES_EQUAL(
            2,
            counters->GetCounter(
                "Availability_TotalIntervals",
                true)->Val());
        UNIT_ASSERT_VALUES_EQUAL(
            1,
            counters->GetCounter(
                "Availability_AvailableIntervals",
                true)->Val());
        UNIT_ASSERT_VALUES_EQUAL(
            1,
            counters->GetCounter(
                "Availability_UnavailableIntervals",
                true)->Val());
        UNIT_ASSERT_VALUES_EQUAL(
            0,
            counters->GetCounter(
                "Availability_LastIntervalAvailable")->Val());
    }
}

}   // namespace NCloud::NFileStore::NFuse