// Отставание читателей: удержание снапшота и наблюдение за lag
// Раздел учебника: Том III, глава 14, «MVCC и снапшоты», §14.5 «RLT».
// Что демонстрирует: read-only транзакция в отдельном потоке удерживает
//   снапшот; коммиты писателя увеличивают её «отставание» (lag);
//   mdbx_reader_list показывает слоты читателей, txnid и lag.
// Сборка: g++ -std=c++17 -I<libmdbx-include> -I../common 19-readers-lag.c++ -lmdbx -pthread
// Запуск: ./19-readers-lag
// Ожидаемый вывод:
//   reader slot: pid=<P> tid=<T> txnid=<N> lag=<L>
//   ok: reader lag observed
// При сбое — сообщение в stderr и exit 1.
//
// SPDX-License-Identifier: Apache-2.0

#include <atomic>
#include <iostream>
#include <string>
#include <thread>

#include "common.h++"
#include <mdbx.h++>

int main() {
  try {
    const std::string path = std::string(example::tmpdir()) + "/mdbx-19-readers-lag.mdbx";
    mdbx::env::remove(path);
    mdbx::env_managed env = example::env_open(path);

    // Несколько коммитов до создания снапшота.
    for (int i = 0; i < 3; ++i) {
      auto txn = env.start_write();
      auto table = txn.open_map(nullptr);
      txn.insert(table, mdbx::slice("k" + std::to_string(i)), mdbx::slice("v"));
      txn.commit();
    }

    // Долгоживущий читатель держит снапшот в отдельном потоке
    // (пересечение read+write на одном потоке запрещено).
    std::atomic<bool> snapshot_ready{false};
    std::atomic<bool> release_reader{false};
    std::thread reader([&] {
      auto rtxn = env.start_read();
      auto table = rtxn.open_map(nullptr);
      (void)rtxn.get(table, mdbx::slice("k2"));
      snapshot_ready = true;
      while (!release_reader.load())
        std::this_thread::yield();
      rtxn.abort();
    });
    while (!snapshot_ready.load())
      std::this_thread::yield();

    // Коммиты после создания снапшота увеличивают lag читателя.
    for (int i = 3; i < 8; ++i) {
      auto txn = env.start_write();
      auto wtable = txn.open_map(nullptr);
      txn.insert(wtable, mdbx::slice("k" + std::to_string(i)), mdbx::slice("v"));
      txn.commit();
    }

    // Перечислить читателей: слот читателя имеет lag > 0.
    struct visitor {
      int operator()(const mdbx::env::reader_info &ri, int) {
        std::cout << "reader slot: pid=" << ri.pid << " tid=" << ri.thread << " txnid=" << ri.transaction_id
                  << " lag=" << ri.transaction_lag << "\n";
        return mdbx::continue_loop;
      }
    } v;
    env.enumerate_readers(v);

    release_reader = true;
    reader.join();

    std::cout << "ok: reader lag observed\n";
    return EXIT_SUCCESS;
  } catch (const std::exception &ex) {
    std::cerr << "error: " << ex.what() << "\n";
    return EXIT_FAILURE;
  }
}