Add a basic OSD test as an example

src/blockstore/blockstore_mock.cpp

@@ -69,17 +69,17 @@ std::string blockstore_mock_t::get_op_diag(blockstore_op_t *op)
 
 uint32_t blockstore_mock_t::get_block_size()
 {
-    return 0;
+    return block_size;
 }
 
 uint64_t blockstore_mock_t::get_block_count()
 {
-    return 0;
+    return block_count;
 }
 
 uint64_t blockstore_mock_t::get_free_block_count()
 {
-    return 0;
+    return block_count;
 }
 
 uint64_t blockstore_mock_t::get_journal_size()
@@ -89,7 +89,7 @@ uint64_t blockstore_mock_t::get_journal_size()
 
 uint32_t blockstore_mock_t::get_bitmap_granularity()
 {
-    return 0;
+    return bitmap_granularity;
 }
 
 uint64_t blockstore_mock_t::get_live_entries()

src/blockstore/blockstore_mock.h

@@ -8,6 +8,9 @@
 class blockstore_mock_t: public blockstore_i
 {
 public:
+    uint32_t block_size = 128*1024;
+    uint32_t bitmap_granularity = 4096;
+    uint64_t block_count = 100*1024*8;
     std::map<uint64_t, uint64_t> inode_space;
 
     blockstore_mock_t(const blockstore_config_t & config);

src/client/etcd_state_client_mock.cpp

@@ -5,6 +5,13 @@
 #include "etcd_state_client_mock.h"
 #include "str_util.h"
 
+etcd_state_client_mock_t::etcd_state_client_mock_t()
+{
+    timespec tv;
+    clock_gettime(CLOCK_REALTIME, &tv);
+    srand48(tv.tv_sec*1000000000 + tv.tv_nsec);
+}
+
 void etcd_state_client_mock_t::etcd_add_watch(json11::Json watch)
 {
 }
@@ -83,6 +90,7 @@ void etcd_state_client_mock_t::etcd_call(std::string api, json11::Json payload,
             else
                 assert(0);
         }
+        std::map<std::string, etcd_kv_t> changes;
         bool has_mod = false;
         for (auto& op: payload[ok ? "success" : "failure"].array_items())
         {
@@ -131,6 +139,14 @@ void etcd_state_client_mock_t::etcd_call(std::string api, json11::Json payload,
                     .mod_revision = mod_revision,
                     .lease_id = lease_id,
                 };
+                std::string err;
+                json11::Json json_value = json11::Json::parse(value, err);
+                if (err != "")
+                {
+                    fprintf(stderr, "Invalid JSON in etcd key %s during test: %s\n", key.c_str(), value.c_str());
+                    exit(1);
+                }
+                changes[key] = { .key = key, .value = json_value, .mod_revision = mod_revision };
                 responses.push_back(json11::Json::object {
                     { "response_put", json11::Json::object{ { "header", json11::Json::object{ { "revision", mod_revision } } } } },
                 });
@@ -143,7 +159,9 @@ void etcd_state_client_mock_t::etcd_call(std::string api, json11::Json payload,
                 uint64_t n_del = 0;
                 for (auto it = data.lower_bound(key); it != data.end() && (range_end == "" || it->first < range_end); )
                 {
-                    printf("\\- del: %s\n", it->first.c_str());
+                    auto & key = it->first;
+                    printf("\\- del: %s\n", key.c_str());
+                    changes[key] = { .key = key, .mod_revision = mod_revision };
                     n_del++;
                     data.erase(it++);
                 }
@@ -152,7 +170,25 @@ void etcd_state_client_mock_t::etcd_call(std::string api, json11::Json payload,
                 });
             }
         }
-        callback("", json11::Json::object{ { "header", json11::Json::object{ { "revision", mod_revision } } }, { "succeeded", ok }, { "responses", responses } });
+        callback("", json11::Json::object{
+            { "header", json11::Json::object{ { "revision", mod_revision } } },
+            { "succeeded", ok },
+            { "responses", responses }
+        });
+        // Push changes to watcher
+        if (changes.size())
+        {
+            for (auto & kv: changes)
+                parse_state(kv.second);
+            if (on_change_hook != NULL)
+                on_change_hook(changes);
+        }
+    }
+    else if (api == "/lease/grant")
+    {
+        uint64_t lease_id = (((uint64_t)lrand48()) << 32) | lrand48();
+        leases[lease_id] = payload["TTL"].uint64_value();
+        callback("", json11::Json::object{ { "ID", std::to_string(lease_id) } });
     }
     else
         callback("Unsupported", json11::Json());

src/client/etcd_state_client_mock.h

@@ -28,7 +28,9 @@ struct etcd_state_client_mock_t: public etcd_state_client_t
     bool paused = false;
     std::vector<etcd_mock_request_t> queue;
 public:
+    std::map<uint64_t, uint64_t> leases;
     std::map<std::string, etcd_mock_key_data_t> data;
+    etcd_state_client_mock_t();
     void set(const std::string& key, json11::Json data, uint64_t mod_revision = 0, uint64_t lease_id = 0);
     void pause();
     void resume();

src/osd/CMakeLists.txt

@@ -39,7 +39,10 @@ target_link_libraries(osd_test
 	vitastor_common
 	Jerasure
 	${ISAL_LIBRARIES}
+	${LIBURING_LIBRARIES}
 )
+add_test(NAME osd_test COMMAND osd_test)
+add_dependencies(build_tests osd_test)
 
 # osd_rmw_test
 add_executable(osd_rmw_test EXCLUDE_FROM_ALL osd_rmw_test.cpp ../util/allocator.cpp)

src/osd/osd.cpp

@@ -30,7 +30,10 @@ osd_t::osd_t(const json11::Json & config, ring_loop_i *ringloop, timerfd_manager
     this->cli_config = config.object_items();
     this->file_config = msgr.read_config(this->cli_config);
     parse_config(true);
+}
 
+void osd_t::start()
+{
     if (json_is_true(this->config["osd_memlock"]))
     {
         // Lock all OSD memory if requested
@@ -130,17 +133,22 @@ void osd_t::init_blockstore(std::function<void()> on_init)
             autosync_writes = max_autosync;
         if (on_init)
         {
-            init_consumer.loop = [this, on_init]()
+            if (bs->is_started())
+                on_init();
+            else
             {
-                // Wait for blockstore initialisation before actually starting OSD logic
-                // to prevent peering timeouts during restart with filled databases
-                if (bs->is_started())
+                init_consumer.loop = [this, on_init]()
                 {
-                    ringloop->set_immediate([this, on_init] { init_consumer.loop = NULL; on_init(); });
-                    ringloop->unregister_consumer(&init_consumer);
-                }
-            };
-            ringloop->register_consumer(&init_consumer);
+                    // Wait for blockstore initialisation before actually starting OSD logic
+                    // to prevent peering timeouts during restart with filled databases
+                    if (bs->is_started())
+                    {
+                        ringloop->set_immediate([this, on_init] { init_consumer.loop = NULL; on_init(); });
+                        ringloop->unregister_consumer(&init_consumer);
+                    }
+                };
+                ringloop->register_consumer(&init_consumer);
+            }
         }
     }
     else if (on_init)

src/osd/osd.h

@@ -97,8 +97,12 @@ struct osd_pg_lock_t
     uint64_t state = 0;
 };
 
+struct osd_test_fixture_t;
+
 class osd_t
 {
+    friend struct osd_test_fixture_t;
+
     // config
 
     json11::Json::object cli_config, file_config, etcd_global_config, etcd_osd_config, config;
@@ -394,6 +398,11 @@ public:
     osd_t(const json11::Json & config, ring_loop_i *ringloop, timerfd_manager_t *tfd, std::unique_ptr<etcd_state_client_t> st_cli_ptr,
         std::function<blockstore_i*(blockstore_config_t & config)> bs_factory);
     ~osd_t();
+    // Wire the OSD into the world: install timers, hook the messenger up to
+    // ringloop, register as a ringloop consumer, and kick off the etcd
+    // config-load chain. Separated from the constructor so tests can build
+    // an inert osd_t, prime mocks, then drive startup explicitly.
+    void start();
     void force_stop(int exitcode);
     bool shutdown();
 };

src/osd/osd_main.cpp

@@ -74,6 +74,7 @@ int main(int narg, char *args[])
             return blockstore_i::create(cfg, ringloop, tfd);
         }
     );
+    osd->start();
     while (1)
     {
         ringloop->loop();

src/test/osd_test.cpp

@@ -3,27 +3,147 @@
 
 #include <stdio.h>
 #include <stdlib.h>
+#include <string.h>
 #include <assert.h>
-#include "osd.h"
-#include "etcd_state_client_mock.h"
-#include "ringloop_mock.h"
-#include "blockstore_mock.h"
 
-void test1()
+#include "osd.h"
+#include "osd_primary.h"
+#include "osd_test_fixture.h"
+
+// Verify that an OSD configured with an etcd address issues a range read for
+// /<prefix>/config/global and /<prefix>/config/pools right after construction.
+//
+// The etcd mock is paused before constructing the OSD, so the txn stays in
+// the mock's queue instead of executing its callback synchronously inside
+// the constructor — that gives us a chance to inspect what the OSD asked
+// for without having to drive the rest of the startup sequence.
+void test_load_global_config()
 {
-    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 *st_cli = new etcd_state_client_mock_t();
-    ring_loop_mock_t *ringloop = new ring_loop_mock_t(RINGLOOP_DEFAULT_SIZE, [&](io_uring_sqe *sqe) {});
-    st_cli->pause();
-    osd_t *osd = new osd_t(config, ringloop, tfd, std::unique_ptr<etcd_state_client_t>(st_cli), [](blockstore_config_t & cfg)
-    {
-        return new blockstore_mock_t({});
+    osd_test_fixture_t f;
+    f.st_cli->pause();
+    f.start(json11::Json::object {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "etcd_prefix", "/vitastor" },
+        { "run_primary", false },
     });
+
+    assert(f.st_cli->queue.size() == 1);
+    assert(f.st_cli->queue[0].api == "/kv/txn");
+    auto & ops = f.st_cli->queue[0].payload["success"].array_items();
+    assert(ops.size() == 2);
+    assert(base64_decode(ops[0]["request_range"]["key"].string_value()) == "/vitastor/config/global");
+    assert(base64_decode(ops[1]["request_range"]["key"].string_value()) == "/vitastor/config/pools");
+
+    printf("test_load_global_config passed\n");
+}
+
+// Build a self-issued OSD_OP_WRITE op for the given inode/offset/length,
+// filled with the given byte. client_id=0 (SELF_CLIENT) so finish_op
+// delivers the reply through the callback we set rather than over the wire.
+static osd_op_t *make_write_op(inode_t inode, uint64_t offset, uint64_t len, uint8_t fill)
+{
+    auto *op = new osd_op_t();
+    op->op_type = OSD_OP_IN;
+    op->client_id = 0;
+    op->req.rw.header.magic = SECONDARY_OSD_OP_MAGIC;
+    op->req.rw.header.id = 1;
+    op->req.rw.header.opcode = OSD_OP_WRITE;
+    op->req.rw.inode = inode;
+    op->req.rw.offset = offset;
+    op->req.rw.len = len;
+    op->buf = malloc(len);
+    memset(op->buf, fill, len);
+    return op;
+}
+
+// Drive a single 4 KiB replicated write through the primary OSD state machine
+// and verify it produces (1) a local blockstore write and (2) a peer subop,
+// then completes the client op once both finish.
+//
+// Layout: pool 1, replicated x2, primary = OSD 1 (us), secondary = OSD 2.
+void test_replicated_write()
+{
+    osd_test_fixture_t f;
+    f.configure_replicated_pool(/*pool_id*/ 1, /*pg_size*/ 2, /*pg_minsize*/ 1, /*pg_count*/ 1,
+        { { 1, 2 } });
+    f.start(json11::Json::object {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "all" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+    });
+    // Peer 2 wasn't online during initial peering -> PG is INCOMPLETE.
+    // Connecting it re-triggers peering; completing the LIST subops with
+    // "no objects" lets the PG transition to PG_ACTIVE.
+    f.connect_peer(2);
+    f.complete_peering_empty();
+    assert(f.pg(1, 1).state & PG_ACTIVE);
+
+    auto *write_op = make_write_op(INODE_WITH_POOL(1, 1), 0, 4096, 0xab);
+    int final_retval = -1;
+    write_op->callback = [&final_retval](osd_op_t *op) {
+        final_retval = op->reply.hdr.retval;
+    };
+    f.exec(write_op);
+
+    // Stage 1: the primary always pumps one subop through SUBMIT_RMW_READ
+    // even for a "fresh" full-block replicated write — for our setup that
+    // subop is a zero-length local read that just resolves the object's
+    // current version.
+    assert(f.bs->queued.size() == 1);
+    auto *zero_read = f.bs->take();
+    assert(zero_read->opcode == BS_OP_READ);
+    assert(zero_read->len == 0);
+    zero_read->version = 0; // object doesn't exist yet -> current version 0
+    zero_read->retval = 0;
+    zero_read->callback(zero_read);
+
+    // Stage 2: with fact_ver=0 and target_ver=1 the primary issues the
+    // actual writes — one local (to bs) and one remote (to OSD 2).
+    auto *peer = f.peer(2);
+    assert(f.bs->queued.size() == 1);
+    assert(peer->sent_ops.size() == 1);
+
+    auto *local_write = f.bs->take();
+    assert(local_write->opcode == BS_OP_WRITE_STABLE);
+    assert(local_write->len == 4096);
+    assert(local_write->oid.inode == INODE_WITH_POOL(1, 1));
+    assert(local_write->oid.stripe == 0);
+    assert(local_write->version == 1);
+
+    auto sent_it = peer->sent_ops.begin();
+    osd_op_t *remote_write = sent_it->second;
+    peer->sent_ops.erase(sent_it);
+    assert(remote_write->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE);
+    assert(remote_write->osd_num == 2);
+    assert(remote_write->req.sec_rw.oid.inode == INODE_WITH_POOL(1, 1));
+    assert(remote_write->req.sec_rw.len == 4096);
+    assert(remote_write->req.sec_rw.version == 1);
+
+    // Local write completes first; the primary is still waiting for the peer.
+    local_write->retval = local_write->len;
+    local_write->callback(local_write);
+    assert(final_retval == -1);
+
+    // Peer reply triggers handle_primary_subop, which sees both subops done
+    // and lets continue_primary_write reach finish_op.
+    remote_write->reply.hdr.retval = remote_write->req.sec_rw.len;
+    remote_write->reply.sec_rw.version = 1;
+    remote_write->callback(remote_write);
+
+    assert(final_retval == 4096);
+    assert(f.pg(1, 1).inflight == 0);
+    assert(f.pg(1, 1).write_queue.empty());
+
+    delete write_op;
+    printf("test_replicated_write passed\n");
 }
 
 int main(int narg, char *args[])
 {
-    test1();
+    test_load_global_config();
+    test_replicated_write();
     return 0;
 }

src/test/osd_test_fixture.h

@@ -0,0 +1,313 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 (see README.md for details)
+
+#pragma once
+
+#include <assert.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "osd.h"
+#include "blockstore_mock.h"
+#include "etcd_state_client_mock.h"
+#include "ringloop_mock.h"
+#include "str_util.h"
+
+// blockstore_mock that captures enqueued ops so tests can drive them by hand.
+class capturing_bs_t: public blockstore_mock_t
+{
+public:
+    using blockstore_mock_t::blockstore_mock_t;
+    std::vector<blockstore_op_t*> queued;
+
+    void enqueue_op(blockstore_op_t *op) override
+    {
+        queued.push_back(op);
+    }
+
+    blockstore_op_t *take()
+    {
+        assert(!queued.empty());
+        auto *op = queued.front();
+        queued.erase(queued.begin());
+        return op;
+    }
+
+    // Pop the first queued op with the given opcode. Aborts if none.
+    blockstore_op_t *take(int opcode)
+    {
+        for (auto it = queued.begin(); it != queued.end(); ++it)
+        {
+            if ((*it)->opcode == opcode)
+            {
+                auto *op = *it;
+                queued.erase(it);
+                return op;
+            }
+        }
+        fprintf(stderr, "capturing_bs_t::take: no queued op with opcode %d\n", opcode);
+        abort();
+    }
+};
+
+// Test harness for primary OSD operations.
+//
+// Owns the mocks (ringloop, timerfd_manager, etcd, blockstore) and exposes
+// helpers that drive the OSD through phases its production lifecycle would
+// otherwise reach via async etcd/network events: configuring pools through
+// the etcd mock, pretending peer OSDs (dis)connected, completing peering
+// LIST subops.
+//
+// A friend of osd_t so it can poke at PG state and message queues; tests
+// themselves should only talk to the fixture's public interface (and to
+// the osd_t for things like exec_op).
+struct osd_test_fixture_t
+{
+    timerfd_manager_t *tfd = nullptr;
+    ring_loop_mock_t *ringloop = nullptr;
+    etcd_state_client_mock_t *st_cli = nullptr; // owned by osd_t after start()
+    osd_t *osd = nullptr;
+    capturing_bs_t *bs = nullptr;
+
+    osd_test_fixture_t()
+    {
+        tfd = new timerfd_manager_t([](int, bool, std::function<void(int, int)>) {});
+        ringloop = new ring_loop_mock_t(RINGLOOP_DEFAULT_SIZE, [](io_uring_sqe *) {});
+        st_cli = new etcd_state_client_mock_t();
+    }
+
+    ~osd_test_fixture_t()
+    {
+        if (osd)
+        {
+            // Drop PGs before tearing msgr down: stop_client would otherwise
+            // call repeer_pgs() on PGs that reference peers we're about to
+            // delete during ~osd_messenger_t().
+            osd->pgs.clear();
+            delete osd;
+        }
+        delete ringloop;
+        delete tfd;
+    }
+
+    // Populate the etcd mock with a single replicated pool's pool/PG config
+    // BEFORE start(). osd_t's constructor-time config load will then pick it
+    // up synchronously and run apply_pg_config() on construction.
+    //
+    // pgs[i] is the osd_set for PG i+1; primary = pgs[i][0].
+    void configure_replicated_pool(pool_id_t pool_id, int pg_size, int pg_minsize,
+        int pg_count, const std::vector<std::vector<osd_num_t>> & pgs)
+    {
+        assert((int)pgs.size() == pg_count);
+        auto pool_id_s = std::to_string(pool_id);
+        st_cli->set("/vitastor/config/pools", json11::Json::object {
+            { pool_id_s, json11::Json::object {
+                { "name", "pool_"+pool_id_s },
+                { "scheme", "replicated" },
+                { "pg_size", pg_size },
+                { "pg_minsize", pg_minsize },
+                { "pg_count", pg_count },
+                { "failure_domain", "osd" },
+                { "immediate_commit", "none" },
+            } },
+        });
+        json11::Json::object items_pool;
+        for (int i = 0; i < pg_count; i++)
+        {
+            json11::Json::array osd_set;
+            for (auto n: pgs[i])
+                osd_set.push_back((double)n);
+            items_pool[std::to_string(i+1)] = json11::Json::object {
+                { "osd_set", osd_set },
+                { "primary", (double)pgs[i][0] },
+            };
+        }
+        st_cli->set("/vitastor/pg/config", json11::Json::object {
+            { "items", json11::Json::object{ { pool_id_s, items_pool } } },
+        });
+    }
+
+    // Construct the osd_t AND call osd->start(). With the etcd mock unpaused,
+    // the whole config-load chain (config/global, config/pools, lease,
+    // /osd/state, pg/config, apply_pg_config, start_pg_peering) runs
+    // synchronously inside start(). Peers that aren't yet connect_peer()'d
+    // will leave their PGs in PG_INCOMPLETE; call connect_peer() +
+    // complete_peering_empty() afterwards to drive them to PG_ACTIVE.
+    //
+    // If a test needs a fully inert osd_t (no timers, no etcd traffic) — for
+    // instance to verify constructor-time behavior — use construct() instead
+    // and call osd->start() yourself.
+    void construct(json11::Json::object osd_config)
+    {
+        assert(!osd);
+        osd = new osd_t(osd_config, ringloop, tfd,
+            std::unique_ptr<etcd_state_client_t>(st_cli),
+            [this](blockstore_config_t & cfg) -> blockstore_i* {
+                return (bs = new capturing_bs_t(cfg));
+            });
+    }
+
+    void start(json11::Json::object osd_config)
+    {
+        construct(osd_config);
+        osd->start();
+    }
+
+    // Pretend a peer OSD has connected (no real TCP). Registers it in the
+    // messenger and triggers re-peering of PGs that include it.
+    void connect_peer(osd_num_t osd_num)
+    {
+        auto *msgr = &osd->msgr;
+        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 disconnect_peer(osd_num_t osd_num)
+    {
+        osd->msgr.stop_client(osd->msgr.osd_peers.at(osd_num)->client_id);
+    }
+
+    // Complete every outstanding peering LIST op (local BS_OP_LIST + peer
+    // OSD_OP_SEC_LIST) with "no objects" so PGs transition to PG_ACTIVE
+    // without us having to invent object lists.
+    void complete_peering_empty()
+    {
+        for (auto it = bs->queued.begin(); it != bs->queued.end(); )
+        {
+            auto *op = *it;
+            if (op->opcode == BS_OP_LIST)
+            {
+                it = bs->queued.erase(it);
+                op->retval = 0;
+                op->version = 0;
+                op->buf = NULL;
+                op->callback(op);
+            }
+            else
+                ++it;
+        }
+        for (auto & p: osd->msgr.osd_peers)
+        {
+            auto *cl = p.second;
+            std::vector<osd_op_t*> list_ops;
+            for (auto & kv: cl->sent_ops)
+                if (kv.second->req.hdr.opcode == OSD_OP_SEC_LIST)
+                    list_ops.push_back(kv.second);
+            for (auto *op: list_ops)
+            {
+                cl->sent_ops.erase(op->req.hdr.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 = 0;
+                op->reply.sec_list.stable_count = 0;
+                op->buf = NULL;
+                op->callback(op);
+            }
+        }
+        // Drive handle_peers so PGs see lists_done and finalize calc_object_states.
+        ringloop->wakeup();
+        ringloop->loop();
+    }
+
+    // Inline list response builder. `objects` are pairs (oid, version);
+    // the first `stable_count` entries are reported as stable. The buffer
+    // is malloc'd and ownership transfers to the peering machinery, which
+    // frees it after calc_object_states.
+    static obj_ver_id *build_list_buf(const std::vector<obj_ver_id> & objects)
+    {
+        if (objects.empty())
+            return nullptr;
+        auto *buf = (obj_ver_id*)malloc(objects.size() * sizeof(obj_ver_id));
+        for (size_t i = 0; i < objects.size(); i++)
+            buf[i] = objects[i];
+        return buf;
+    }
+
+    // Pop the first BS_OP_LIST from bs->queued and reply with the given
+    // object list. The peering callback consumes (and later frees) op->buf.
+    void reply_local_list(const std::vector<obj_ver_id> & objects, uint64_t stable_count)
+    {
+        auto *op = bs->take(BS_OP_LIST);
+        op->buf = (uint8_t*)build_list_buf(objects);
+        op->retval = (int)objects.size();
+        op->version = stable_count;
+        op->callback(op);
+    }
+
+    // Pop the OSD_OP_SEC_LIST sent to `osd_num` and reply with the given
+    // object list. Same ownership rules as reply_local_list.
+    void reply_peer_list(osd_num_t osd_num,
+        const std::vector<obj_ver_id> & objects, uint64_t stable_count)
+    {
+        auto *cl = osd->msgr.osd_peers.at(osd_num);
+        osd_op_t *op = nullptr;
+        for (auto & kv: cl->sent_ops)
+        {
+            if (kv.second->req.hdr.opcode == OSD_OP_SEC_LIST)
+            {
+                op = kv.second;
+                break;
+            }
+        }
+        assert(op);
+        cl->sent_ops.erase(op->req.hdr.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 = (int64_t)objects.size();
+        op->reply.sec_list.stable_count = stable_count;
+        op->buf = build_list_buf(objects);
+        op->callback(op);
+    }
+
+    // Pop the first sent op of given opcode from peer's outbox. Aborts if
+    // none — caller should know which subops to expect.
+    osd_op_t *peer_take(osd_num_t osd_num, uint64_t opcode)
+    {
+        auto *cl = osd->msgr.osd_peers.at(osd_num);
+        for (auto & kv: cl->sent_ops)
+        {
+            if (kv.second->req.hdr.opcode == opcode)
+            {
+                auto *op = kv.second;
+                cl->sent_ops.erase(op->req.hdr.id);
+                return op;
+            }
+        }
+        fprintf(stderr, "peer_take: OSD %ju has no sent op with opcode %ju\n", osd_num, opcode);
+        abort();
+    }
+
+    // Drive ringloop to flush handle_peers and any pending lambdas.
+    void pump()
+    {
+        ringloop->wakeup();
+        ringloop->loop();
+    }
+
+    // Submit an op to the OSD's primary state machine. Wrapper because
+    // osd_t::exec_op is private — tests aren't friends, the fixture is.
+    void exec(osd_op_t *op)
+    {
+        osd->exec_op(op);
+    }
+
+    osd_client_t *peer(osd_num_t osd_num)
+    {
+        return osd->msgr.osd_peers.at(osd_num);
+    }
+
+    pg_t &pg(pool_id_t pool, pg_num_t num)
+    {
+        return osd->pgs.at({ .pool_id = pool, .pg_num = num });
+    }
+};

src/test/ringloop_mock.h

@@ -4,6 +4,7 @@
 #pragma once
 
 #include "ringloop.h"
+#include <map>
 
 class ring_loop_mock_t: public ring_loop_i
 {

src/test/test_cluster_client.cpp

@@ -114,24 +114,34 @@ void check_completed(int *r)
     delete r;
 }
 
-void pretend_connected(cluster_client_t *cli, osd_num_t osd_num)
+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 = cli->msgr.next_client_id++;
+    cl->client_id = msgr->next_client_id++;
     cl->osd_num = osd_num;
     cl->peer_fd = -1;
     cl->peer_state = PEER_CONNECTED;
-    cli->msgr.osd_peers[osd_num] = cl;
-    cli->msgr.clients[cl->client_id] = cl;
-    cli->msgr.wanted_peers.erase(osd_num);
-    cli->msgr.repeer_pgs(osd_num);
+    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)
 {
-    printf("OSD %ju disconnected\n", osd_num);
-    cli->msgr.stop_client(cli->msgr.osd_peers.at(osd_num)->client_id);
+    pretend_disconnected(&cli->msgr, osd_num);
 }
 
 void check_disconnected(cluster_client_t *cli, osd_num_t osd_num)