// Readers lag: holding a snapshot and observing the lag
// Textbook section: Volume III, chapter 14, "MVCC and snapshots", §14.5 "RLT".
// What it demonstrates: a read-only transaction in a separate thread holds
//   a snapshot; writer commits increase its "lag";
//   mdbx_reader_list shows reader slots, txnid and lag.
// Build: g++ -std=c++17 -I<libmdbx-include> -I../common 19-readers-lag.c++ -lmdbx -pthread
// Run: ./19-readers-lag
// Expected output:
//   reader slot: pid=<P> tid=<T> txnid=<N> lag=<L>
//   ok: reader lag observed
// On failure — message to stderr and 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);

    // A few commits before the snapshot is created.
    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();
    }

    // A long-lived reader holds the snapshot in a separate thread
    // (mixing read and write on one thread is forbidden).
    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();

    // Commits after the snapshot creation increase the reader's 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();
    }

    // Enumerate the readers: a reader slot has 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;
  }
}