// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)

#ifdef WITH_OPENSSL
#include <openssl/rand.h>
#endif

#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include "etcd_state_client_mock.h"
#include "cluster_client_impl.h"
#include "msgr_encrypt.h"

class cluster_client_test_t
{
public:
    static void continue_ops(cluster_client_t *cli)
    {
        cli->continue_ops(cli->client_retry_interval);
    }
};

void configure_single_pg_pool(etcd_state_client_mock_t *mock)
{
    mock->set("/vitastor/config/pools", json11::Json::object {
        { "1", json11::Json::object {
            { "name", "hddpool" },
            { "scheme", "replicated" },
            { "pg_size", 2 },
            { "pg_minsize", 1 },
            { "pg_count", 1 },
            { "failure_domain", "osd" },
            { "immediate_commit", "none" },
        } }
    });
    mock->set("/vitastor/pg/config", json11::Json::object {
        { "items", json11::Json::object {
            { "1", json11::Json::object {
                { "1", json11::Json::object {
                    { "osd_set", json11::Json::array { 1, 2 } },
                    { "primary", 1 },
                } }
            } }
        } }
    });
    mock->set("/vitastor/pg/state/1/1", json11::Json::object {
        { "peers", json11::Json::array { 1, 2 } },
        { "primary", 1 },
        { "state", json11::Json::array { "active" } },
    });
}

int *test_write(cluster_client_t *cli, uint64_t offset, uint64_t len, uint8_t c, std::function<void()> cb = NULL, bool instant = false)
{
    printf("Post write %jx+%jx\n", offset, len);
    int *r = new int;
    *r = instant ? -2 : -1;
    cluster_op_t *op = new cluster_op_t();
    op->opcode = OSD_OP_WRITE;
    op->inode = 0x1000000000001;
    op->offset = offset;
    op->len = len;
    op->iov.push_back(malloc_or_die(len), len);
    memset(op->iov.buf[0].iov_base, c, len);
    op->callback = [r, cb](cluster_op_t *op)
    {
        if (*r == -1)
            printf("Error: Not allowed to complete yet (retval %d)\n", op->retval);
        assert(*r != -1);
        *r = op->retval == op->len ? 1 : 0;
        free(op->iov.buf[0].iov_base);
        printf("Done write %jx+%jx r=%d\n", op->offset, op->len, op->retval);
        delete op;
        if (cb != NULL)
            cb();
    };
    cli->execute(op);
    if (instant)
    {
        long res = *r;
        assert(*r >= 0);
        delete r;
        return (int*)res;
    }
    return r;
}

int *test_sync(cluster_client_t *cli)
{
    printf("Post sync\n");
    int *r = new int;
    *r = -1;
    cluster_op_t *op = new cluster_op_t();
    op->opcode = OSD_OP_SYNC;
    op->callback = [r](cluster_op_t *op)
    {
        if (*r == -1)
            printf("Error: Not allowed to complete yet\n");
        assert(*r != -1);
        *r = op->retval == 0 ? 1 : 0;
        printf("Done sync r=%d\n", op->retval);
        delete op;
    };
    cli->execute(op);
    return r;
}

void can_complete(int *r)
{
    // Allow the operation to proceed so the test verifies
    // that it doesn't complete earlier than expected
    *r = -2;
}

#define check_completed(r) { assert(*(r) == 1); delete (r); }

void pretend_connected(osd_messenger_t *msgr, osd_num_t osd_num)
{
    printf("OSD %ju connected\n", osd_num);
    auto cl = new osd_client_t();
    cl->client_id = msgr->next_client_id++;
    cl->osd_num = osd_num;
    cl->peer_fd = -1;
    cl->peer_state = PEER_CONNECTED;
    msgr->osd_peers[osd_num] = cl;
    msgr->clients[cl->client_id] = cl;
    msgr->wanted_peers.erase(osd_num);
    msgr->repeer_pgs(osd_num);
}

void pretend_disconnected(osd_messenger_t *msgr, osd_num_t osd_num)
{
    printf("OSD %ju disconnected\n", osd_num);
    msgr->stop_client(msgr->osd_peers.at(osd_num)->client_id);
}

void pretend_connected(cluster_client_t *cli, osd_num_t osd_num)
{
    pretend_connected(&cli->msgr, osd_num);
}

void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num)
{
    pretend_disconnected(&cli->msgr, osd_num);
}

void check_disconnected(cluster_client_t *cli, osd_num_t osd_num)
{
    if (cli->msgr.osd_peers.find(osd_num) != cli->msgr.osd_peers.end())
    {
        printf("OSD %ju not disconnected as it ought to be\n", osd_num);
        assert(0);
    }
}

void check_op_count(cluster_client_t *cli, osd_num_t osd_num, int ops)
{
    osd_client_t *cl = cli->msgr.osd_peers.at(osd_num);
    int real_ops = cl->sent_ops.size();
    if (real_ops != ops)
    {
        printf("error: %d ops expected, but %d queued\n", ops, real_ops);
        assert(0);
    }
}

osd_op_t *find_op(cluster_client_t *cli, osd_num_t osd_num, uint64_t opcode, uint64_t offset, uint64_t len)
{
    osd_client_t *cl = cli->msgr.osd_peers.at(osd_num);
    auto op_it = cl->sent_ops.begin();
    while (op_it != cl->sent_ops.end())
    {
        auto op = op_it->second;
        if (op->req.hdr.opcode == opcode && (opcode == OSD_OP_SYNC ||
            op->req.rw.inode == 0x1000000000001 && op->req.rw.offset == offset && op->req.rw.len == len))
        {
            return op;
        }
        op_it++;
    }
    op_it = cl->sent_ops.begin();
    while (op_it != cl->sent_ops.end())
    {
        printf("Found opcode %ju offset %jx size %x\n", op_it->second->req.hdr.opcode, op_it->second->req.rw.offset, op_it->second->req.rw.len);
        op_it++;
    }
    printf("Not found opcode %ju offset %jx size %jx\n", opcode, offset, len);
    return NULL;
}

void pretend_op_completed(cluster_client_t *cli, osd_op_t *op, int64_t retval)
{
    assert(op);
    printf("Pretend completed %s %jx+%x\n", op->req.hdr.opcode == OSD_OP_SYNC
        ? "sync" : (op->req.hdr.opcode == OSD_OP_WRITE ? "write" : "read"), op->req.rw.offset, op->req.rw.len);
    uint64_t op_id = op->req.hdr.id;
    uint64_t client_id = op->client_id;
    cli->msgr.clients[client_id]->sent_ops.erase(op_id);
    op->reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
    op->reply.hdr.id = op->req.hdr.id;
    op->reply.hdr.opcode = op->req.hdr.opcode;
    op->reply.hdr.retval = retval < 0 ? retval : (op->req.hdr.opcode == OSD_OP_SYNC ? 0 : op->req.rw.len);
    // Copy lambda to be unaffected by `delete op`
    std::function<void(osd_op_t*)>(op->callback)(op);
}

void test1()
{
    json11::Json config;
    timerfd_manager_t *tfd = new timerfd_manager_t([](int fd, bool wr, std::function<void(int, int)> callback){});
    etcd_state_client_mock_t *mock = new etcd_state_client_mock_t();
    mock->pause();
    cluster_client_t *cli = new cluster_client_t(NULL, tfd, config, std::unique_ptr<etcd_state_client_t>(mock));

    int *r1 = test_write(cli, 0, 4096, 0x55);
    configure_single_pg_pool(mock);
    mock->resume();
    pretend_connected(cli, 1);
    can_complete(r1);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 4096), 0);
    check_completed(r1);
    r1 = test_write(cli, 4096, 4096, 0x56);
    can_complete(r1);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 4096, 4096), 0);
    check_completed(r1);
    pretend_disconnected(cli, 1);
    int *r2 = test_sync(cli);
    pretend_connected(cli, 1);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 8192), 0);
    check_op_count(cli, 1, 1);
    can_complete(r2);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
    check_completed(r2);
    // Check that the client doesn't repeat operations once more
    pretend_disconnected(cli, 1);
    pretend_connected(cli, 1);
    check_op_count(cli, 1, 0);

    // Case:
    // Write(1) -> Complete Write(1) -> Overwrite(2) -> Complete Write(2)
    // -> Overwrite(3) -> Drop OSD connection -> Reestablish OSD connection
    // -> Complete All Posted Writes -> Sync -> Complete Sync
    // The resulting state of the block must be (3) over (2) over (1).
    // I.e. the part overwritten by (3) must remain as in (3) and so on.

    // More interesting case:
    // Same, but both Write(2) and Write(3) must consist of two parts:
    // one from an OSD 2 that drops connection and other from OSD 1 that doesn't.
    // The idea is that if the whole Write(2) is repeated when OSD 2 drops connection
    // then it may also overwrite a part in OSD 1 which shouldn't be overwritten.

    // Another interesting case:
    // A new operation added during replay (would also break with the previous implementation)

    r1 = test_write(cli, 0, 0x10000, 0x56);
    can_complete(r1);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 0x10000), 0);
    check_completed(r1);

    r1 = test_write(cli, 0xE000, 0x4000, 0x57);
    can_complete(r1);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0xE000, 0x4000), 0);
    check_completed(r1);

    r1 = test_write(cli, 0x10000, 0x4000, 0x58);

    pretend_disconnected(cli, 1);
    pretend_connected(cli, 1);
    cluster_client_test_t::continue_ops(cli);

    // Check replay
    {
        uint64_t replay_start = UINT64_MAX;
        uint64_t replay_end = 0;
        std::vector<osd_op_t*> replay_ops;
        auto osd_cl = cli->msgr.osd_peers.at(1);
        for (auto & op_p: osd_cl->sent_ops)
        {
            auto op = op_p.second;
            assert(op->req.hdr.opcode == OSD_OP_WRITE);
            uint64_t offset = op->req.rw.offset;
            if (op->req.rw.offset < replay_start)
                replay_start = op->req.rw.offset;
            if (op->req.rw.offset+op->req.rw.len > replay_end)
                replay_end = op->req.rw.offset+op->req.rw.len;
            for (int buf_idx = 0; buf_idx < op->iov.count; buf_idx++)
            {
                for (int i = 0; i < op->iov.buf[buf_idx].iov_len; i++, offset++)
                {
                    uint8_t c = offset < 0xE000 ? 0x56 : (offset < 0x10000 ? 0x57 : 0x58);
                    if (((uint8_t*)op->iov.buf[buf_idx].iov_base)[i] != c)
                    {
                        printf("Write replay: mismatch at %ju (expected %02x, have %02x)\n", offset-op->req.rw.offset,
                            c, ((uint8_t*)op->iov.buf[buf_idx].iov_base)[i]);
                        goto fail;
                    }
                }
            }
        fail:
            assert(offset == op->req.rw.offset+op->req.rw.len);
            replay_ops.push_back(op);
        }
        if (replay_start != 0 || replay_end != 0x10000)
        {
            printf("Write replay: range mismatch: 0x%jx-0x%jx (expected 0-0x10000)\n", replay_start, replay_end);
            assert(0);
        }
        for (auto op: replay_ops)
        {
            pretend_op_completed(cli, op, 0);
        }
    }
    // Check that the following write finally proceeds
    check_op_count(cli, 1, 1);
    can_complete(r1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0x10000, 0x4000), 0);
    check_completed(r1);
    check_op_count(cli, 1, 0);

    // Check sync
    r2 = test_sync(cli);
    can_complete(r2);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
    check_completed(r2);

    // Check disconnect during write
    r1 = test_write(cli, 0, 4096, 0x59);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 0x1000), -EPIPE);
    check_disconnected(cli, 1);
    pretend_connected(cli, 1);
    cluster_client_test_t::continue_ops(cli);
    check_op_count(cli, 1, 1);
    can_complete(r1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 0x1000), 0);
    check_completed(r1);

    // Check disconnect inside operation callback (reenterability)
    // Probably doesn't happen too often, but possible in theory
    r1 = test_write(cli, 0, 0x1000, 0x60, [cli]()
    {
        pretend_disconnected(cli, 1);
    });
    r2 = test_write(cli, 0x1000, 0x1000, 0x61);
    check_op_count(cli, 1, 2);
    can_complete(r1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 0x1000), 0);
    check_completed(r1);
    check_disconnected(cli, 1);
    pretend_connected(cli, 1);
    cluster_client_test_t::continue_ops(cli);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 0x1000), 0);
    check_op_count(cli, 1, 1);
    can_complete(r2);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0x1000, 0x1000), 0);
    check_completed(r2);

    // Free client
    delete cli;
    delete tfd;
    printf("[ok] write replay test\n");
}

void test2()
{
    writeback_cache_t *wb = new writeback_cache_t();
    cluster_op_t *op = new cluster_op_t();
    op->opcode = OSD_OP_WRITE;
    op->inode = 1;
    op->offset = 0;
    op->len = 4096;
    op->iov.push_back(malloc_or_die(4096*1024), 4096);
    // 0-4k = 0x55
    memset(op->iov.buf[0].iov_base, 0x55, op->iov.buf[0].iov_len);
    wb->copy_write(op, CACHE_WRITTEN);
    // 8k-12k = 0x66
    op->offset = 8192;
    memset(op->iov.buf[0].iov_base, 0x66, op->iov.buf[0].iov_len);
    wb->copy_write(op, CACHE_WRITTEN);
    // 4k-1M+4k = 0x77
    op->len = op->iov.buf[0].iov_len = 1048576;
    op->offset = 4096;
    memset(op->iov.buf[0].iov_base, 0x77, op->iov.buf[0].iov_len);
    wb->copy_write(op, CACHE_WRITTEN);
    // check it
    assert(wb->dirty_buffers.size() == 2);
    auto uit = wb->dirty_buffers.begin();
    int i;
    assert(uit->first.inode == 1);
    assert(uit->first.stripe == 0);
    assert(uit->second.len == 4096);
    for (i = 0; i < uit->second.len && ((uint8_t*)uit->second.buf)[i] == 0x55; i++) {}
    assert(i == uit->second.len);
    uit++;
    assert(uit->first.inode == 1);
    assert(uit->first.stripe == 4096);
    assert(uit->second.len == 1048576);
    for (i = 0; i < uit->second.len && ((uint8_t*)uit->second.buf)[i] == 0x77; i++) {}
    assert(i == uit->second.len);
    uit++;
    // free memory
    free(op->iov.buf[0].iov_base);
    delete op;
    delete wb;
    printf("[ok] copy_write test\n");
}

void test_writeback()
{
    json11::Json config = json11::Json::object {
        { "client_enable_writeback", true },
        { "client_writeback_allowed", true },
        { "client_max_buffered_bytes", 1024*1024 },
        { "client_max_buffered_ops", 2 },
        { "client_max_writeback_iodepth", 2 },
        { "client_max_dirty_bytes", 1024*1024 },
        { "client_max_dirty_ops", 2 },
    };
    timerfd_manager_t *tfd = new timerfd_manager_t([](int fd, bool wr, std::function<void(int, int)> callback){});
    etcd_state_client_mock_t *mock = new etcd_state_client_mock_t();
    mock->pause();
    cluster_client_t *cli = new cluster_client_t(NULL, tfd, config, std::unique_ptr<etcd_state_client_t>(mock));

    configure_single_pg_pool(mock);
    mock->resume();
    pretend_connected(cli, 1);

    // Check that 3 consecutive writes are merged by writeback
    assert((long)test_write(cli, 0, 4096, 0x55, NULL, true) == 1);
    check_op_count(cli, 1, 0);
    assert((long)test_write(cli, 4096, 4096, 0x55, NULL, true) == 1);
    check_op_count(cli, 1, 0);
    assert((long)test_write(cli, 8192, 4096, 0x55, NULL, true) == 1);
    check_op_count(cli, 1, 0);

    assert((long)test_write(cli, 1024*1024, 4096, 0x66, NULL, true) == 1);
    check_op_count(cli, 1, 0);

    // 3rd and 4th writes should trigger 1 writeback each
    assert((long)test_write(cli, 2*1024*1024, 4096, 0x66, NULL, true) == 1);
    check_op_count(cli, 1, 1);
    assert((long)test_write(cli, 3*1024*1024, 4096, 0x66, NULL, true) == 1);
    check_op_count(cli, 1, 2);

    // 5th write should be postponed until at least 1 writeback is completed
    int *r1 = test_write(cli, 4*1024*1024, 4096, 0x67, NULL);
    check_op_count(cli, 1, 2);
    can_complete(r1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 3*4096), 0);
    check_completed(r1);
    // autosync because max_dirty_ops=2, flush waits for sync
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 1024*1024, 4096), 0);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 2*1024*1024, 4096), 0);
    check_op_count(cli, 1, 0);

    int *r2 = test_sync(cli);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 3*1024*1024, 4096), 0);
    check_op_count(cli, 1, 1);
    // autosync because max_dirty_ops=2, flush waits for sync
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 4*1024*1024, 4096), 0);
    check_op_count(cli, 1, 1);
    can_complete(r2);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
    check_completed(r2);

    // Check cutting of the beginning and end
    assert((long)test_write(cli, 0, 32768, 0x55, NULL, true) == 1);
    check_op_count(cli, 1, 0);
    assert((long)test_write(cli, 32768, 32768, 0x56, NULL, true) == 1);
    check_op_count(cli, 1, 0);
    assert((long)test_write(cli, 16384, 32768, 0x57, NULL, true) == 1);
    check_op_count(cli, 1, 0);
    assert((long)test_write(cli, 16384+4096, 32768-4096, 0x58, NULL, true) == 1);
    check_op_count(cli, 1, 0);
    r2 = test_sync(cli);
    check_op_count(cli, 1, 1);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 65536), 0);
    check_op_count(cli, 1, 1);
    can_complete(r2);
    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
    check_completed(r2);

    // Free client
    delete cli;
    delete tfd;
    printf("[ok] writeback test\n");
}

static void copy_write_for_test(writeback_cache_t *wb, uint64_t offset, uint64_t len, int state, uint64_t new_flush_id)
{
    void *buf = malloc_or_die(len);
    cluster_op_t *op = new cluster_op_t();
    op->opcode = OSD_OP_WRITE;
    op->inode = 0x1000000000001;
    op->offset = offset;
    op->len = len;
    op->iov.push_back(buf, len);
    wb->copy_write(op, state, new_flush_id);
    delete op;
    free(buf);
}

void test_writeback_merge()
{
    writeback_cache_t *wb = new writeback_cache_t;
    // [1000..3000]
    copy_write_for_test(wb, 1000, 2000, CACHE_DIRTY, 0);
    assert(wb->writeback_bytes == 2000);
    assert(wb->writeback_queue_size == 1);
    // [1000..3000][3000..4000]
    copy_write_for_test(wb, 3000, 1000, CACHE_DIRTY, 0);
    assert(wb->writeback_bytes == 3000);
    assert(wb->writeback_queue_size == 1);
    // [1000..3000][3000..4000], [5000..6000]
    copy_write_for_test(wb, 5000, 1000, CACHE_DIRTY, 0);
    assert(wb->writeback_bytes == 4000);
    assert(wb->writeback_queue_size == 2);
    // [1000..2500], [3500..4000], [5000..6000]
    copy_write_for_test(wb, 2500, 1000, CACHE_WRITTEN, 0);
    assert(wb->writeback_bytes == 3000);
    assert(wb->writeback_queue_size == 3);
    // [1000..2500], [3500..4000][4000...5000][5000..6000]
    copy_write_for_test(wb, 4000, 1000, CACHE_DIRTY, 0);
    assert(wb->writeback_bytes == 4000);
    assert(wb->writeback_queue_size == 2);
    // [1000..2500], [3500..4500][4500...5000][5000..6000]
    copy_write_for_test(wb, 3500, 1000, CACHE_DIRTY, 0);
    assert(wb->writeback_bytes == 4000);
    assert(wb->writeback_queue_size == 2);
    // [1000..2500], [3500..4500], [5000..6000]
    copy_write_for_test(wb, 4500, 500, CACHE_WRITTEN, 0);
    assert(wb->writeback_bytes == 3500);
    assert(wb->writeback_queue_size == 3);
    // [1000..2500][2500..3500][3500..4500], [5000..6000]
    copy_write_for_test(wb, 2500, 1000, CACHE_DIRTY, 0);
    assert(wb->writeback_bytes == 4500);
    assert(wb->writeback_queue_size == 2);
    // [1000..2500][2500..3500][3500..4500], [5500..6000]
    copy_write_for_test(wb, 5000, 500, CACHE_WRITTEN, 0);
    assert(wb->writeback_bytes == 4000);
    assert(wb->writeback_queue_size == 2);
    // [1000..2500][2500..3500][3500..4000], [5500..6000]
    copy_write_for_test(wb, 4000, 1000, CACHE_WRITTEN, 0);
    assert(wb->writeback_bytes == 3500);
    assert(wb->writeback_queue_size == 2);
    // [1000..2500][2500..3500][3500..4000][4000..5500][5500..6000]
    copy_write_for_test(wb, 4000, 1500, CACHE_DIRTY, 0);
    assert(wb->writeback_bytes == 5000);
    assert(wb->writeback_queue_size == 1);
    delete wb;
    printf("[ok] writeback merge test\n");
}

// Verifies that writeback_queue contains an entry for each sequence of
// CACHE_DIRTY buffers, otherwise flushing never reaches it
static void check_writeback_queued(writeback_cache_t *wb)
{
    int runs = 0;
    int not_queued = 0;
    auto it = wb->dirty_buffers.begin();
    while (it != wb->dirty_buffers.end())
    {
        if (it->second.state != CACHE_DIRTY)
        {
            it++;
            continue;
        }
        auto run_start = it;
        uint64_t expected_next = run_start->first.stripe;
        bool expected_is_del = (run_start->second.buf == NULL);
        auto run_end = it;
        while (run_end != wb->dirty_buffers.end() &&
            run_end->second.state == CACHE_DIRTY &&
            run_end->first.inode == run_start->first.inode &&
            run_end->first.stripe == expected_next &&
            (run_end->second.buf == NULL) == expected_is_del)
        {
            expected_next = run_end->first.stripe + run_end->second.len;
            run_end++;
        }
        runs++;
        bool queued = false;
        for (auto & a: wb->writeback_queue)
        {
            if (a.inode == run_start->first.inode &&
                a.stripe >= run_start->first.stripe &&
                a.stripe < expected_next)
            {
                queued = true;
                break;
            }
        }
        if (!queued)
        {
            printf("FAIL: CACHE_DIRTY inode %jx range %ju-%ju is missing from the writeback queue\n",
                run_start->first.inode, run_start->first.stripe, expected_next);
            not_queued++;
        }
        it = run_end;
    }
    assert(!not_queued);
    assert(wb->writeback_queue_size == runs);
}

void test_writeback_queue_split()
{
    // Regression test for the future case of client_enable_writeback disabled at runtime
    writeback_cache_t *wb = new writeback_cache_t;
    // CACHE_DIRTY [1000..5000] -- buffered with writeback on
    copy_write_for_test(wb, 1000, 4000, CACHE_DIRTY, 0);
    assert(wb->writeback_queue_size == 1);
    check_writeback_queued(wb);
    // CACHE_DIRTY [10000..14000] -- buffered with writeback on
    copy_write_for_test(wb, 10000, 4000, CACHE_DIRTY, 0);
    assert(wb->writeback_queue_size == 2);
    check_writeback_queued(wb);
    // CACHE_DIRTY [20000..24000] -- buffered with writeback on
    copy_write_for_test(wb, 20000, 4000, CACHE_DIRTY, 0);
    assert(wb->writeback_queue_size == 3);
    check_writeback_queued(wb);
    // Writeback is now off; a CACHE_REPEATING write at [2000..3000] splits the run:
    // [1000..2000] CACHE_DIRTY, [2000..3000] CACHE_REPEATING, [3000..5000] CACHE_DIRTY
    // The right half [3000..5000] must have its own anchor in writeback_queue.
    copy_write_for_test(wb, 2000, 1000, CACHE_REPEATING, 1);
    assert(wb->dirty_buffers.size() == 5);
    assert(wb->writeback_queue_size == 4);
    check_writeback_queued(wb);
    // Second split - only the end part is left
    copy_write_for_test(wb, 10000, 1000, CACHE_REPEATING, 1);
    assert(wb->dirty_buffers.size() == 6);
    assert(wb->writeback_queue_size == 4);
    check_writeback_queued(wb);
    // Third split - only the beginning part is left
    copy_write_for_test(wb, 23000, 1000, CACHE_REPEATING, 1);
    assert(wb->dirty_buffers.size() == 7);
    assert(wb->writeback_queue_size == 4);
    check_writeback_queued(wb);
    delete wb;
    printf("[ok] test_writeback_queue_split\n");
}

#ifdef WITH_OPENSSL
void test_msgr_encrypt()
{
    const size_t sz = 1048576;
    uint8_t *src = (uint8_t*)malloc_or_die(sz);
    for (size_t i = 0; i < sz; i++)
        src[i] = (i*0x1001) % 256;
    uint8_t *crypt = (uint8_t*)malloc_or_die(sz);
    uint8_t *decrypt = (uint8_t*)malloc_or_die(sz);
    uint8_t *crypt2 = (uint8_t*)malloc_or_die(sz);
    uint8_t *key = (uint8_t*)malloc_or_die(64);
    RAND_bytes(key, 64);

    // Basic encrypt+decrypt and also get reference data

    auto enc = new op_aes_xts_encrypt_t();
    enc->start(key, 4096 * 113, 4096);
    size_t in_pos = 0;
    size_t out_pos = 0;
    while (out_pos < sz)
    {
        enc->update(src+in_pos, sz-in_pos, crypt+out_pos, sz-out_pos, in_pos, out_pos);
    }

    auto dec = new op_aes_xts_decrypt_t();
    dec->start(&key, 1, NULL, 4096 * 113, 4096);
    in_pos = out_pos = 0;
    while (out_pos < sz)
    {
        dec->update(crypt+in_pos, sz-in_pos, decrypt+out_pos, sz-out_pos, in_pos, out_pos);
    }

    assert(memcmp(src, decrypt, sz) == 0);

    // Insufficient output encrypt
    printf("...insufficient output encrypt\n");
    enc->start(key, 4096 * 114, 4096);
    in_pos = out_pos = 0;
    enc->update(src+4096, 4096, crypt2, 4095, in_pos, out_pos);
    assert(in_pos == 4096);
    assert(out_pos == 4095);
    enc->update(src+4096+in_pos, 4096-in_pos, crypt2+out_pos, 4096-out_pos, in_pos, out_pos);
    assert(in_pos == 4096);
    assert(out_pos == 4096);
    assert(memcmp(crypt2, crypt+4096, 4096) == 0);

    // Fragmented encrypt
    printf("...fragmented encrypt\n");
    enc->start(key, 4096 * 114, 4096);
    in_pos = out_pos = 0;
    enc->update(src+4096, 2000, crypt2, 4095, in_pos, out_pos);
    assert(in_pos == 2000);
    assert(out_pos == 0);
    enc->update(src+4096+in_pos, 4096-in_pos, crypt2+out_pos, 4096-out_pos, in_pos, out_pos);
    assert(in_pos == 4096);
    assert(out_pos == 4096);
    assert(memcmp(crypt2, crypt+4096, 4096) == 0);

    // Fragmented decrypt
    // Input: 1000 + 2000 + 3000 + 2192, output: 500 + 3000 + 1000 + 3000 + 692
    printf("...fragmented decrypt\n");
    dec->start(&key, 1, NULL, 4096 * 114, 4096);
    in_pos = out_pos = 0;
    dec->update(crypt+4096, 1000, decrypt, 500, in_pos, out_pos);
    assert(in_pos == 1000);
    assert(out_pos == 0);
    dec->update(crypt+4096+1000, 2000, decrypt, 500, in_pos, out_pos);
    assert(in_pos == 3000);
    assert(out_pos == 0);
    dec->update(crypt+4096+3000, 3000, decrypt, 500, in_pos, out_pos);
    assert(in_pos == 4096);
    assert(out_pos == 500);
    dec->update(crypt+4096+in_pos, 6000-in_pos, decrypt+out_pos, 3000, in_pos, out_pos);
    assert(in_pos == 4096);
    assert(out_pos == 3500);
    dec->update(crypt+4096+in_pos, 6000-in_pos, decrypt+out_pos, 1000, in_pos, out_pos);
    assert(in_pos == 4096);
    assert(out_pos == 4096);
    dec->update(crypt+4096+in_pos, 6000-in_pos, decrypt+out_pos, 4500-out_pos, in_pos, out_pos);
    assert(in_pos == 6000);
    assert(out_pos == 4096);
    dec->update(crypt+4096+in_pos, 8192-in_pos, decrypt+out_pos, 4500-out_pos, in_pos, out_pos);
    assert(in_pos == 8192);
    assert(out_pos == 4500);
    dec->update(crypt+4096+in_pos, 8192-in_pos, decrypt+out_pos, 7500-out_pos, in_pos, out_pos);
    assert(in_pos == 8192);
    assert(out_pos == 7500);
    dec->update(crypt+4096+in_pos, 8192-in_pos, decrypt+out_pos, 8192-out_pos, in_pos, out_pos);
    assert(in_pos == 8192);
    assert(out_pos == 8192);
    assert(memcmp(decrypt, src+4096, 8192) == 0);

    // Extra size decrypt
    // Input: 8192, output: 4096
    printf("...extra size decrypt\n");
    dec->start(&key, 1, NULL, 4096 * 114, 4096);
    in_pos = out_pos = 0;
    dec->update(crypt+4096, 8192, decrypt, 4096, in_pos, out_pos);
    assert(in_pos == 4096);
    assert(out_pos == 4096);
    assert(memcmp(decrypt, src+4096, 4096) == 0);

    delete dec;
    delete enc;

    free(key);
    free(crypt2);
    free(decrypt);
    free(crypt);
    free(src);
    printf("[ok] msgr aes-xts encryption test\n");
}

void test_msgr_decrypt_chain()
{
    const size_t sz = 4096 * 4;
    uint8_t *src = (uint8_t*)malloc_or_die(sz);
    for (size_t i = 0; i < sz; i++)
        src[i] = (i*0x1001) % 256;
    uint8_t *crypt = (uint8_t*)malloc_or_die(sz);
    uint8_t *decrypt = (uint8_t*)malloc_or_die(sz);
    uint8_t *key = (uint8_t*)malloc_or_die(64);
    RAND_bytes(key, 64);
    uint8_t *key2 = (uint8_t*)malloc_or_die(64);
    RAND_bytes(key2, 64);

    // Chained decryption with multiple keys

    // encrypt:
    size_t in_pos = 0, out_pos = 0;
    auto enc = new op_aes_xts_encrypt_t();
    // block 1 with key1
    enc->start(key, 4096 * 113, 4096);
    enc->update(src, 4096, crypt, 4096, in_pos, out_pos);
    assert(in_pos == 4096 && out_pos == 4096);
    // block 2 as plain
    memcpy(crypt + 4096, src + 4096, 4096);
    // block 3 with key2
    enc->start(key2, 4096 * 115, 4096);
    enc->update(src + 2*4096, 4096, crypt + 2*4096, 4096, in_pos, out_pos);
    assert(in_pos == 2*4096 && out_pos == 2*4096);
    // block 4 again with key1
    enc->start(key, 4096 * 116, 4096);
    enc->update(src + 3*4096, 4096, crypt + 3*4096, 4096, in_pos, out_pos);
    assert(in_pos == 3*4096 && out_pos == 3*4096);

    // decrypt:
    uint8_t* keys[3] = { key, key2, NULL };
    uint8_t chain_info[4] = { 0, 2, 1, 0 };
    auto dec = new op_aes_xts_decrypt_t();
    dec->start(keys, 3, chain_info, 4096 * 113, 4096);
    in_pos = out_pos = 0;
    while (out_pos < sz)
    {
        dec->update(crypt+in_pos, sz-in_pos, decrypt+out_pos, sz-out_pos, in_pos, out_pos);
    }

    assert(memcmp(src, decrypt, sz) == 0);

    delete dec;
    delete enc;

    free(key2);
    free(key);
    free(decrypt);
    free(crypt);
    free(src);
    printf("[ok] msgr aes-xts chained decrypt\n");
}
#endif

int main(int narg, char *args[])
{
    test1();
    test2();
    test_writeback();
    test_writeback_merge();
    test_writeback_queue_split();
#ifdef WITH_OPENSSL
    test_msgr_encrypt();
    test_msgr_decrypt_chain();
#endif
    return 0;
}