/*

Copyright (c) 2024-2026, Arvid Norberg
All rights reserved.

You may use, distribute and modify this code under the terms of the BSD license,
see LICENSE file.
*/

#include "libtorrent/aux_/visit_block_iovecs.hpp"
#include "libtorrent/hasher.hpp"
#include <array>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <chrono>
#include "test.hpp"
#include "test_utils.hpp"
#include "disk_cache_test_utils.hpp"

using lt::span;
using namespace lt;
using namespace lt::aux;

namespace {

struct tbe
{
	span<char const> _buf;
};

// found via ADL by visit_block_iovecs when iterating a span<tbe const>
span<char const> write_buf(tbe const& be) { return be._buf; }

template <size_t N>
tbe b(char const (&literal)[N])
{
	auto buf = span<char const>{&literal[0], N - 1};
	return tbe{buf};
}

std::string join(span<span<char const>> iovec)
{
	std::string ret;
	for (span<char const> const& b : iovec)
	{
		ret.append(b.begin(), b.end());
	}
	return ret;
}

}

TORRENT_TEST(visit_block_iovecs_full)
{
	std::array<tbe, 5> const blocks{b("a"), b("b"), b("c"), b("d"), b("e")};

	int cnt = 0;
	lt::aux::visit_block_iovecs(span<tbe const>(blocks)
		, [&cnt] (span<span<char const>> iovec, int start_idx) {
		TEST_EQUAL(cnt, 0);
		TEST_EQUAL(start_idx, 0);
		TEST_EQUAL(iovec.size(), 5);
		TEST_EQUAL(join(iovec), "abcde");
		++cnt;
		return false;
	});
}

TORRENT_TEST(visit_block_iovecs_one_hole)
{
	std::array<tbe, 5> const blocks{b("a"), b("b"), b(""), b("d"), b("e")};

	int cnt = 0;
	lt::aux::visit_block_iovecs(span<tbe const>(blocks)
		, [&cnt] (span<span<char const>> iovec, int start_idx) {
		switch (cnt) {
			case 0:
				TEST_EQUAL(start_idx, 0);
				TEST_EQUAL(iovec.size(), 2);
				TEST_EQUAL(join(iovec), "ab");
				break;
			case 1:
				TEST_EQUAL(start_idx, 3);
				TEST_EQUAL(iovec.size(), 2);
				TEST_EQUAL(join(iovec), "de");
				break;
			default:
				TORRENT_ASSERT_FAIL();
		}
		++cnt;
		return false;
	});
}

TORRENT_TEST(visit_block_iovecs_two_holes)
{
	std::array<tbe, 5> const blocks{b("a"), b(""), b("c"), b(""), b("e")};

	int cnt = 0;
	lt::aux::visit_block_iovecs(span<tbe const>(blocks)
		, [&cnt] (span<span<char const>> iovec, int start_idx) {
		switch (cnt) {
			case 0:
				TEST_EQUAL(start_idx, 0);
				TEST_EQUAL(iovec.size(), 1);
				TEST_EQUAL(join(iovec), "a");
				break;
			case 1:
				TEST_EQUAL(start_idx, 2);
				TEST_EQUAL(iovec.size(), 1);
				TEST_EQUAL(join(iovec), "c");
				break;
			case 2:
				TEST_EQUAL(start_idx, 4);
				TEST_EQUAL(iovec.size(), 1);
				TEST_EQUAL(join(iovec), "e");
				break;
			default:
				TORRENT_ASSERT_FAIL();
		}
		++cnt;
		return false;
	});
}


TORRENT_TEST(visit_block_iovecs_interrupt)
{
	std::array<tbe, 3> const blocks{b("a"), b(""), b("c")};

	int cnt = 0;
	lt::aux::visit_block_iovecs(span<tbe const>(blocks)
		, [&cnt] (span<span<char const>> iovec, int start_idx) {
		switch (cnt) {
			case 0:
				TEST_EQUAL(start_idx, 0);
				TEST_EQUAL(iovec.size(), 1);
				TEST_EQUAL(join(iovec), "a");
				break;
			default:
				TORRENT_ASSERT_FAIL();
		}
		++cnt;
		return true;
	});
}

TORRENT_TEST(visit_block_iovecs_leading_hole)
{
	std::array<tbe, 5> const blocks{b(""), b("a"), b("b"), b("c"), b("d")};

	int cnt = 0;
	lt::aux::visit_block_iovecs(span<tbe const>(blocks)
		, [&cnt] (span<span<char const>> iovec, int start_idx) {
		TEST_EQUAL(cnt, 0);
		TEST_EQUAL(start_idx, 1);
		TEST_EQUAL(iovec.size(), 4);
		TEST_EQUAL(join(iovec), "abcd");
		++cnt;
		return false;
	});
}

TORRENT_TEST(visit_block_iovecs_trailing_hole)
{
	std::array<tbe, 5> const blocks{b("a"), b("b"), b("c"), b("d"), b("")};

	int cnt = 0;
	lt::aux::visit_block_iovecs(span<tbe const>(blocks)
		, [&cnt] (span<span<char const>> iovec, int start_idx) {
		TEST_EQUAL(cnt, 0);
		TEST_EQUAL(start_idx, 0);
		TEST_EQUAL(iovec.size(), 4);
		TEST_EQUAL(join(iovec), "abcd");
		++cnt;
		return false;
	});
}

namespace {

// all blocks arrive and are hashed before flush
void test_disk_bottleneck(test_mode_t const mode)
{
	cache_fixture f(2, mode);

	auto r0 = f.insert(0_piece, 0);
	// need_hasher_kick is set for v1/hybrid pieces when the v1 piece hasher
	// can advance (this insert filled hasher_cursor's slot). Pure-v2 pieces
	// don't drive the cache hasher; pread_disk_io wakes the hasher
	// independently for v2 storages.
	if (mode & test_mode::v1)
		TEST_CHECK(bool(r0 & disk_cache::need_hasher_kick));
	else
		TEST_CHECK(!(r0 & disk_cache::need_hasher_kick));

	auto r1 = f.insert(0_piece, 1);
	// needs_hasher_kick_flag already set from block 0 (v1/hybrid case) or
	// never set (pure v2 case) — either way not raised here.
	TEST_CHECK(!(r1 & disk_cache::need_hasher_kick));

	TEST_EQUAL(int(f.cache.size()), 2);

	// kick_hashers() advances v1 and also drains the v2 hash queue,
	// capturing computed v2 hashes in f.v2_hashes.
	f.kick_hashers();

	std::vector<sha256_hash> block_hashes(mode & test_mode::v2 ? 2 : 0);
	auto hash_job = std::make_unique<pread_disk_job>();
	hash_job->action = job::hash{
		{}, 0_piece,
		span<sha256_hash>{block_hashes.data(), int(block_hashes.size())},
		sha1_hash{}
	};

	auto const hpr = f.cache.try_hash_piece(f.loc(0_piece), hash_job.get());
	if (mode & test_mode::v1)
	{
		// v1 (or hybrid) -- the piece hasher advanced to blocks_in_piece in
		// kick_hashers(), so the piece hash is ready.
		TEST_EQUAL(hpr, disk_cache::hash_result::job_completed);
		TEST_CHECK(!std::get<job::hash>(hash_job->action).piece_hash.is_all_zeros());
	}
	else
	{
		// Pure-v2 -- no v1 piece hasher, so hasher_cursor never reaches
		// blocks_in_piece and try_hash_piece returns post_job. The actual
		// path in production is pread_disk_io::do_job(hash), which pulls
		// the v2 hashes from precomputed_block_hashes.
		TEST_EQUAL(hpr, disk_cache::hash_result::post_job);
	}
	if (mode & test_mode::v2)
	{
		// v2 block hashes live on the storage's precomputed_block_hashes
		// (captured into f.v2_hashes by the test fixture's kick_hashers()).
		TEST_EQUAL(int(f.v2_hashes.size()), 2);
		TEST_CHECK(!f.v2_hashes.at({0_piece, 0}).is_all_zeros());
		TEST_CHECK(!f.v2_hashes.at({0_piece, 1}).is_all_zeros());
	}

	TEST_EQUAL(f.flush(), 2);
	TEST_EQUAL(int(f.cache.size()), 0);
}

// block 0 flushed before the hasher runs — try_hash_piece returns post_job
void test_hashing_bottleneck(test_mode_t const mode)
{
	cache_fixture f(2, mode);

	f.insert(0_piece, 0);
	TEST_EQUAL(f.flush(0), 1);
	f.insert(0_piece, 1);

	std::vector<sha256_hash> block_hashes(mode & test_mode::v2 ? 2 : 0);
	auto hash_job = std::make_unique<pread_disk_job>();
	hash_job->action = job::hash{
		{}, 0_piece,
		span<sha256_hash>{block_hashes.data(), int(block_hashes.size())},
		sha1_hash{}
	};

	// v2-only with v2_pending > 0: try_hash_piece hangs the hash_job on the
	// cpe so drain_v2_hash_queue can post it once the queue drains, and
	// clear_piece_impl extracts the hung hash_job alongside block 1's
	// write_job. v1/hybrid don't take that gate -- kick_hasher hasn't run
	// so the v1_done check sends them down post_job.
	bool const v2_only = (mode & test_mode::v2) && !(mode & test_mode::v1);
	TEST_EQUAL(f.cache.try_hash_piece(f.loc(0_piece), hash_job.get()),
		v2_only ? disk_cache::hash_result::job_queued : disk_cache::hash_result::post_job);

	jobqueue_t aborted;
	TEST_CHECK(f.cache.try_clear_piece(f.loc(0_piece), nullptr, aborted));
	TEST_EQUAL(aborted.size(), v2_only ? 2 : 1);
	TEST_EQUAL(int(f.cache.size()), 0);
}

// three independent pieces: each hashed and flushed independently
void test_multi_piece(test_mode_t const mode)
{
	cache_fixture f(1, mode);

	f.insert(0_piece, 0);
	f.insert(1_piece, 0);
	f.insert(2_piece, 0);
	TEST_EQUAL(int(f.cache.size()), 3);

	f.kick_hashers();

	for (auto p : {0_piece, 1_piece, 2_piece})
	{
		sha256_hash bh;
		auto hash_job = std::make_unique<pread_disk_job>();
		hash_job->action = job::hash{
			{}, p,
			(mode & test_mode::v2) ? span<sha256_hash>{&bh, 1} : span<sha256_hash>{},
			sha1_hash{}
		};
		auto const hpr = f.cache.try_hash_piece(f.loc(p), hash_job.get());
		if (mode & test_mode::v1)
		{
			TEST_EQUAL(hpr, disk_cache::hash_result::job_completed);
			TEST_CHECK(!std::get<job::hash>(hash_job->action).piece_hash.is_all_zeros());
		}
		else
		{
			// Pure-v2: hasher_cursor doesn't advance, so try_hash_piece
			// returns post_job (handled by pread_disk_io::do_job(hash) in
			// production).
			TEST_EQUAL(hpr, disk_cache::hash_result::post_job);
		}
		if (mode & test_mode::v2)
		{
			// v2 hashes are stashed on storage via drain_v2_hash_queue;
			// the test fixture captured them in v2_hashes.
			TEST_CHECK(!f.v2_hashes.at({p, 0}).is_all_zeros());
		}
	}

	TEST_EQUAL(f.flush(), 3);
	TEST_EQUAL(int(f.cache.size()), 0);
}

}

TORRENT_TEST(disk_bottleneck_v1) { test_disk_bottleneck(test_mode::v1); }
TORRENT_TEST(disk_bottleneck_v2) { test_disk_bottleneck(test_mode::v2); }
TORRENT_TEST(disk_bottleneck_hybrid) { test_disk_bottleneck(test_mode::v1 | test_mode::v2); }

TORRENT_TEST(hashing_bottleneck_v1) { test_hashing_bottleneck(test_mode::v1); }
TORRENT_TEST(hashing_bottleneck_v2) { test_hashing_bottleneck(test_mode::v2); }
TORRENT_TEST(hashing_bottleneck_hybrid) { test_hashing_bottleneck(test_mode::v1 | test_mode::v2); }

TORRENT_TEST(multi_piece_v1) { test_multi_piece(test_mode::v1); }
TORRENT_TEST(multi_piece_v2) { test_multi_piece(test_mode::v2); }
TORRENT_TEST(multi_piece_hybrid) { test_multi_piece(test_mode::v1 | test_mode::v2); }

// v2, hashing is the bottleneck: block 0 flushed before SHA256 is computed.
// hash2() must invoke the fallback (read from disk).
TORRENT_TEST(v2_hashing_bottleneck)
{
	// 1-block piece for simplicity.
	cache_fixture f(1, test_mode::v2);

	f.insert(0_piece, 0);
	// Flush before the hasher runs -- block_hashes[0] stays all-zeros.
	TEST_EQUAL(f.flush(0), 1);

	// The precomputed hash is absent and the buffer is gone; fallback fires.
	bool fallback_called = false;
	sha256_hash h = f.cache.hash2(f.loc(0_piece), 0, [&]() -> sha256_hash {
		fallback_called = true;
		return sha256_hash{};
	});
	TEST_CHECK(fallback_called);
	(void)h;

	jobqueue_t aborted;
	TEST_CHECK(f.cache.try_clear_piece(f.loc(0_piece), nullptr, aborted));
	TEST_CHECK(aborted.empty()); // block 0's write_job was consumed by the flush
	TEST_EQUAL(int(f.cache.size()), 0);
	TEST_EQUAL(f.alloc.live, 0); // buffer was freed when the block was flushed
}

// v2: hash2() served from precomputed in-cache hash (no fallback needed).
TORRENT_TEST(v2_hash2_from_cache)
{
	cache_fixture f(1, test_mode::v2);

	f.insert(0_piece, 0);

	// Let the hasher compute SHA256 for block 0.
	jobqueue_t completed, retry;
	f.cache.kick_pending_hashers(completed, retry);

	bool fallback_called = false;
	sha256_hash h = f.cache.hash2(f.loc(0_piece), 0, [&]() -> sha256_hash {
		fallback_called = true;
		return sha256_hash{};
	});
	TEST_CHECK(!fallback_called);
	TEST_CHECK(!h.is_all_zeros());

	jobqueue_t aborted;
	TEST_CHECK(f.cache.try_clear_piece(f.loc(0_piece), nullptr, aborted));
	TEST_EQUAL(aborted.size(), 1); // block 0's write_job was never flushed
	TEST_EQUAL(int(f.cache.size()), 0);
}

// Clear a piece that was never flushed or hashed.
// All write_jobs must be returned in the aborted queue.
TORRENT_TEST(clear_piece_v1)
{
	cache_fixture f(2, test_mode::v1);

	f.insert(0_piece, 0);
	f.insert(0_piece, 1);
	TEST_EQUAL(int(f.cache.size()), 2);

	jobqueue_t aborted;
	TEST_CHECK(f.cache.try_clear_piece(f.loc(0_piece), nullptr, aborted));
	TEST_EQUAL(int(f.cache.size()), 0);
	TEST_EQUAL(aborted.size(), 2);
}

// Clear a piece that is not in the cache — returns true immediately.
TORRENT_TEST(clear_piece_not_in_cache)
{
	cache_fixture f(1, test_mode::v1);

	jobqueue_t aborted;
	TEST_CHECK(f.cache.try_clear_piece(f.loc(0_piece), nullptr, aborted));
	TEST_CHECK(aborted.empty());
	TEST_EQUAL(int(f.cache.size()), 0);
}

// Hash failure: piece is hashed in-cache, hash check fails, piece is cleared
// before flushing. Both write_jobs must appear in the aborted queue.
TORRENT_TEST(hash_failure_clear)
{
	cache_fixture f(2, test_mode::v1);

	f.insert(0_piece, 0);
	f.insert(0_piece, 1);

	jobqueue_t completed, retry;
	f.cache.kick_pending_hashers(completed, retry);

	auto hash_job = std::make_unique<pread_disk_job>();
	hash_job->action = job::hash{{}, 0_piece, span<sha256_hash>{}, sha1_hash{}};
	TEST_EQUAL(f.cache.try_hash_piece(f.loc(0_piece), hash_job.get())
		, disk_cache::hash_result::job_completed);

	// Simulate hash mismatch: clear the piece before it reaches disk.
	jobqueue_t aborted;
	TEST_CHECK(f.cache.try_clear_piece(f.loc(0_piece), nullptr, aborted));
	TEST_EQUAL(aborted.size(), 2);
	TEST_EQUAL(int(f.cache.size()), 0);
}

// Clear a partially-flushed piece.
// Block 0 is hashed then flushed (its write_job is consumed by the flush).
// Block 1 is only in cache. Only block 1's write_job is aborted.
TORRENT_TEST(clear_piece_partially_flushed)
{
	cache_fixture f(2, test_mode::v1);

	// Insert and hash block 0 so hasher_cursor advances to 1.
	f.insert(0_piece, 0);
	{
		jobqueue_t completed, retry;
		f.cache.kick_pending_hashers(completed, retry);
	}

	// Flush block 0 (cheap: already hashed).
	f.flush(0);

	// Insert block 1.
	f.insert(0_piece, 1);
	TEST_EQUAL(int(f.cache.size()), 1); // only block 1's buffer remains

	jobqueue_t aborted;
	TEST_CHECK(f.cache.try_clear_piece(f.loc(0_piece), nullptr, aborted));
	TEST_EQUAL(int(f.cache.size()), 0);
	TEST_EQUAL(aborted.size(), 1); // only block 1's write_job
}

namespace {

	// Pure-v2 / hybrid variant of clear_piece_v1: verifies that
	// clear_piece_impl's m_v2_hash_queue dedup loop drops every queued entry
	// for the piece and decrements v2_pending to zero. Both inserts left their
	// buffers in m_v2_hash_queue, so size() before clear is queue-driven.
	void test_clear_piece_queued_v2(test_mode_t const mode)
	{
		cache_fixture f(2, mode);

		f.insert(0_piece, 0);
		f.insert(0_piece, 1);
		TEST_EQUAL(int(f.cache.size()), 2); // both entries in m_v2_hash_queue

		jobqueue_t aborted;
		TEST_CHECK(f.cache.try_clear_piece(f.loc(0_piece), nullptr, aborted));
		TEST_EQUAL(int(f.cache.size()), 0); // queue dedup'd both entries
		TEST_EQUAL(aborted.size(), 2); // both write_jobs aborted
		TEST_EQUAL(f.alloc.live, 0);
	}

}

TORRENT_TEST(clear_piece_queued_v2) { test_clear_piece_queued_v2(test_mode::v2); }
TORRENT_TEST(clear_piece_queued_hybrid)
{
	test_clear_piece_queued_v2(test_mode::v1 | test_mode::v2);
}

namespace {

	// Branch 2: clear_piece arrives while flushing_flag is set. The flush
	// callback runs with the cache mutex released; try_clear_piece must
	// park the clear_piece job on the cpe and return false. flush_piece_impl's
	// post-scope_end block then dispatches the parked clear via
	// clear_piece_fun once flushing_flag is cleared.
	void test_clear_piece_during_flush(test_mode_t const mode)
	{
		cache_fixture f(2, mode);
		f.insert(0_piece, 0);
		f.insert(0_piece, 1);

		auto clear_job = std::make_unique<pread_disk_job>();
		clear_job->action = job::clear_piece{{}, 0_piece};

		jobqueue_t cb_aborted;
		bool cb_immediate = true;
		int callback_calls = 0;

		jobqueue_t deferred_aborted;
		disk_job* deferred_clear = nullptr;

		f.cache.flush_to_disk(
			[&](bitfield&, span<disk_job* const>) -> int {
				++callback_calls;
				// cache mutex is released while we're in here; the cpe has
				// flushing_flag set, so try_clear_piece parks our clear_job
				// rather than running clear_piece_impl inline.
				cb_immediate = f.cache.try_clear_piece(f.loc(0_piece), clear_job.get(), cb_aborted);
				// Don't flush anything: leave the write_jobs to be aborted by
				// the deferred clear.
				return 0;
			},
			0, // target=0, optimistic=false -> reaches the expensive pass
			[&](jobqueue_t aborted, disk_job* clear) {
				while (!aborted.empty())
					deferred_aborted.push_back(aborted.pop_front());
				if (clear) deferred_clear = clear;
			},
			false);

		TEST_EQUAL(callback_calls, 1);
		TEST_CHECK(!cb_immediate); // try_clear_piece deferred
		TEST_CHECK(cb_aborted.empty()); // nothing returned through the inline path
		TEST_EQUAL(deferred_aborted.size(), 2); // both write_jobs aborted via clear_piece_fun
		TEST_EQUAL(deferred_clear, clear_job.get());
		TEST_EQUAL(int(f.cache.size()), 0);
		// alloc.live is mode-dependent here: v1 keeps the buffers attached to
		// the aborted write_jobs (released only when the caller processes them);
		// v2 moved the buffers into m_v2_hash_queue at insert time and clear_piece_impl's
		// queue dedup drops them. So we don't assert on it.
	}

}

TORRENT_TEST(clear_piece_during_flush_v1) { test_clear_piece_during_flush(test_mode::v1); }
TORRENT_TEST(clear_piece_during_flush_v2) { test_clear_piece_during_flush(test_mode::v2); }
TORRENT_TEST(clear_piece_during_flush_hybrid)
{
	test_clear_piece_during_flush(test_mode::v1 | test_mode::v2);
}

namespace {

	// Branch 5: clear_piece arrives while a drain batch holds the only
	// outstanding v2_pending decrement. clear_piece_impl drops queued entries
	// (none -- already popped by drain) without decrementing v2_pending; the
	// post-clear_piece_impl check in try_clear_piece sees v2_pending > 0 and
	// parks the clear_job. When the drain re-acquires the lock and decrements
	// v2_pending to zero, its second batch loop dispatches the parked clear
	// via clear_piece_fun.
	void test_clear_piece_v2_drain_in_flight(test_mode_t const mode)
	{
		cache_fixture f(1, mode);
		f.insert(0_piece, 0);
		TEST_EQUAL(int(f.cache.size()), 1); // one queue entry

		auto clear_job = std::make_unique<pread_disk_job>();
		clear_job->action = job::clear_piece{{}, 0_piece};

		jobqueue_t cb_aborted;
		bool cb_immediate = true;
		int store_calls = 0;

		jobqueue_t deferred_aborted;
		disk_job* deferred_clear = nullptr;

		jobqueue_t posted;
		f.cache.drain_v2_hash_queue(
			[&](std::shared_ptr<aux::pread_storage> const&,
				piece_index_t,
				int,
				sha256_hash const&) {
				++store_calls;
				// We're inside the unlocked window of drain_v2_hash_queue:
				// the entry has been popped from m_v2_hash_queue but
				// v2_pending has not yet been decremented. try_clear_piece
				// must defer -- v2_pending stays > 0 until drain re-acquires
				// the lock.
				cb_immediate = f.cache.try_clear_piece(f.loc(0_piece), clear_job.get(), cb_aborted);
			},
			posted,
			[&](jobqueue_t aborted, disk_job* clear) {
				while (!aborted.empty())
					deferred_aborted.push_back(aborted.pop_front());
				if (clear) deferred_clear = clear;
			});

		TEST_EQUAL(store_calls, 1);
		TEST_CHECK(!cb_immediate);
		// clear_piece_impl ran inside try_clear_piece, so the write_job came
		// back through the inline aborted queue, not the deferred one.
		TEST_EQUAL(cb_aborted.size(), 1);
		TEST_CHECK(deferred_aborted.empty());
		TEST_EQUAL(deferred_clear, clear_job.get());
		TEST_EQUAL(int(f.cache.size()), 0);
		TEST_EQUAL(f.alloc.live, 0);
	}

}

TORRENT_TEST(clear_piece_v2_drain_in_flight_v2)
{
	test_clear_piece_v2_drain_in_flight(test_mode::v2);
}
TORRENT_TEST(clear_piece_v2_drain_in_flight_hybrid)
{
	test_clear_piece_v2_drain_in_flight(test_mode::v1 | test_mode::v2);
}

namespace {

	// Same scenario as clear_piece_v2_drain_in_flight, but exercised through the
	// fixture's full kick_hashers() pipeline. Verifies the fixture's capture
	// fields (cleared_jobs, aborted_jobs) reflect the deferred-clear dispatch
	// the drain made on the way out.
	void test_deferred_clear_via_kick_hashers(test_mode_t const mode)
	{
		cache_fixture f(1, mode);
		f.insert(0_piece, 0);

		auto clear_job = std::make_unique<pread_disk_job>();
		clear_job->action = job::clear_piece{{}, 0_piece};

		jobqueue_t cb_aborted;
		bool cb_immediate = true;
		f.drain_store_hook = [&] {
			// inside the drain's unlocked window, v2_pending is still > 0 (the
			// entry has been popped but not yet decremented), so try_clear_piece
			// parks the clear and dispatch is left to the drain's second pass.
			cb_immediate = f.cache.try_clear_piece(f.loc(0_piece), clear_job.get(), cb_aborted);
		};

		f.kick_hashers();

		TEST_CHECK(!cb_immediate);
		TEST_EQUAL(cb_aborted.size(), 1); // write_job, aborted inline by clear_piece_impl
		TEST_EQUAL(f.cleared_jobs.size(), 1); // dispatched by the drain's clear_piece_fun
		TEST_EQUAL(f.cleared_jobs[0], clear_job.get());
		TEST_CHECK(f.aborted_jobs.empty()); // drain passes empty aborted to clear_piece_fun
		TEST_EQUAL(int(f.cache.size()), 0);
		TEST_EQUAL(f.alloc.live, 0);
	}

}

TORRENT_TEST(deferred_clear_via_kick_hashers_v2)
{
	test_deferred_clear_via_kick_hashers(test_mode::v2);
}
TORRENT_TEST(deferred_clear_via_kick_hashers_hybrid)
{
	test_deferred_clear_via_kick_hashers(test_mode::v1 | test_mode::v2);
}

// try_hash_piece's v2-pending defer branch parks the hash_job on the cpe;
// once the drain decrements v2_pending to zero, its second pass extracts the
// hash_job and posts it via the `posted` jobqueue. Pure-v2 only -- the
// hybrid path is covered by hashing_bottleneck_hybrid via a different gate.
TORRENT_TEST(parked_hash_job_dispatched_via_drain_v2)
{
	cache_fixture f(1, test_mode::v2);
	f.insert(0_piece, 0);

	sha256_hash bh;
	auto hash_job = std::make_unique<pread_disk_job>();
	hash_job->action = job::hash{{}, 0_piece, span<sha256_hash>{&bh, 1}, sha1_hash{}};

	// no v1 hasher cursor + v2_pending > 0 -> try_hash_piece parks the
	// hash_job on the cpe and returns job_queued.
	TEST_EQUAL(f.cache.try_hash_piece(f.loc(0_piece), hash_job.get()),
		disk_cache::hash_result::job_queued);

	// kick_hashers drains the queue and dispatches the parked hash_job via
	// drain_v2_hash_queue's `posted` parameter.
	f.kick_hashers();

	TEST_EQUAL(f.posted_jobs.size(), 1);
	TEST_CHECK(f.posted_jobs.pop_front() == hash_job.get());
	// the queue entry has migrated back to the cbe (its buffer was handed
	// back to wj.buf), so the block sits in the cache as a dirty write
	// awaiting flush.
	TEST_EQUAL(int(f.cache.size()), 1);
	TEST_EQUAL(f.v2_hashes.size(), 1u);
	TEST_CHECK(!f.v2_hashes.at({0_piece, 0}).is_all_zeros());
	TEST_EQUAL(f.flush(), 1);
	TEST_EQUAL(int(f.cache.size()), 0);
}

// flush_piece_impl releases the cache mutex during the flush callback.
// Inserts into previously-empty slots of the piece being flushed are legal
// at that point (insert() only requires block_idx >= hasher_cursor). The
// callback must observe a snapshot of write_jobs taken before the lock was
// released, so concurrent inserts cannot race with the flush.
TORRENT_TEST(insert_during_flush_snapshot_stable)
{
	cache_fixture f(4, test_mode::v1);

	// Populate blocks 0 and 1. Leave 2 and 3 empty.
	f.insert(0_piece, 0);
	f.insert(0_piece, 1);
	TEST_EQUAL(int(f.cache.size()), 2);

	int callback_calls = 0;
	f.cache.flush_to_disk(
		[&](bitfield& flushed, span<disk_job* const> blocks) -> int {
			++callback_calls;
			TEST_EQUAL(blocks.size(), 4);
			TEST_CHECK(blocks[0] != nullptr);
			TEST_CHECK(blocks[1] != nullptr);
			TEST_CHECK(blocks[2] == nullptr);
			TEST_CHECK(blocks[3] == nullptr);

			// The cache mutex is released while this callback runs. Stand in
			// for a concurrent network thread by inserting a new block at
			// index 2. The snapshot was taken under the mutex, so what we
			// already hold must not reflect this insertion.
			f.insert(0_piece, 2);
			TEST_CHECK(blocks[2] == nullptr);

			// Flush only what was in the snapshot.
			flushed.set_bit(0);
			flushed.set_bit(1);
			return 2;
		},
		0, // target=0: triggers the expensive flush pass (full piece span)
		[](jobqueue_t, disk_job*) {},
		false);

	TEST_EQUAL(callback_calls, 1);
	// Blocks 0 and 1 are flushed. Block 2 (inserted during the flush) is
	// still in the cache as a dirty block; the flush did not touch it.
	TEST_EQUAL(int(f.cache.size()), 1);
}

// A piece can have piece_hash_returned_flag set with some block slots still
// in monostate. This happens when the user resumes a partially-complete
// piece: only the missing blocks go through async_write, the rest stay on
// disk. When the bittorrent layer later asks for the piece hash, hash_piece()
// reads the on-disk blocks itself, sets piece_hash_returned_flag, and leaves
// the piece in the cache because flushed_cursor < blocks_in_piece.
//
// flush_storage() (run on torrent teardown) must be able to free that piece
// without tripping the precondition in free_piece(). Prior to the fix the
// "piece_hash_returned_flag implies flushed_cursor == blocks_in_piece"
// assertion fired here.
TORRENT_TEST(flush_storage_after_hash_piece_with_monostate_block)
{
	cache_fixture f(2, test_mode::v1);

	// Simulate a partial-resume: only block 1 is written this session.
	// Block 0 is on disk from a previous session, so its slot stays in
	// monostate.
	f.insert(0_piece, 1);
	TEST_EQUAL(int(f.cache.size()), 1);

	// The hasher cannot advance past slot 0 (no buffer), so hasher_cursor
	// stays at 0 and try_hash_piece() falls through to post_job.
	auto hash_job = std::make_unique<pread_disk_job>();
	hash_job->action = job::hash{{}, 0_piece, span<sha256_hash>{}, sha1_hash{}};
	TEST_EQUAL(
		f.cache.try_hash_piece(f.loc(0_piece), hash_job.get()), disk_cache::hash_result::post_job);

	// Drive hash_piece() the same way pread_disk_io::do_job(hash) does: the
	// callback represents reading any nullptr-blocks back from disk to
	// finish the SHA-1. We only need it to be called for block 0 (the
	// monostate slot) -- the actual hash value is irrelevant for this test.
	bool callback_invoked = false;
	auto const res = f.cache.hash_piece(f.loc(0_piece),
		hash_job.get(),
		[&](aux::piece_hasher*, std::uint16_t const hasher_cursor, span<char const*> const blocks) {
			callback_invoked = true;
			TEST_EQUAL(hasher_cursor, 0);
			TEST_EQUAL(blocks.size(), 2);
			TEST_CHECK(blocks[0] == nullptr); // would be read from disk
			TEST_CHECK(blocks[1] != nullptr);
		});
	TEST_CHECK(res == disk_cache::hash_piece_result::completed);
	TEST_CHECK(callback_invoked);

	// flushed_cursor < blocks_in_piece here (slot 0 is still monostate), so
	// the piece is left in cache by hash_piece(). flush_storage() must
	// flush block 1 and then free the piece without tripping the assertion.
	f.flush_storage_for();

	TEST_EQUAL(int(f.cache.size()), 0);
	TEST_EQUAL(f.alloc.live, 0);
}

// Two threads calling flush_storage() concurrently for the same storage must
// not both wait on the same in-flight flush of a piece. notify_flushed_flag
// records only that *a* waiter exists, not which thread, so a second waiter
// would be woken by the same notify, race the first to erase the piece, and
// dereference a now-dangling piece_iter. The fix makes the second
// flush_storage() observe notify_flushed_flag and skip the piece.
//
// Reproduction: a background flush_to_disk() holds flushing_flag on the piece,
// parked inside its flush callback with the cache mutex released. Two
// flush_storage() threads then race; exactly one sets notify_flushed_flag and
// waits, the other observes the flag and skips. Once the skipper returns we
// know the waiter is parked: it set the flag under the cache mutex and then
// cv-waited (atomically releasing the lock the skipper went on to take). Only
// then do we release the background flush, so the waiter flushes and erases
// the piece.
TORRENT_TEST(flush_storage_concurrent_same_storage)
{
	cache_fixture f(2, test_mode::v1);
	f.insert(0_piece, 0);
	f.insert(0_piece, 1);
	TEST_EQUAL(int(f.cache.size()), 2);

	std::mutex mtx;
	std::condition_variable cv;
	bool bg_parked = false;
	bool release_bg = false;
	int finished = 0;

	// Background flush: parks inside the callback while flushing_flag is set on
	// piece 0 and the cache mutex is released. Flushes nothing -- the waiting
	// flush_storage() flushes the blocks once it wakes.
	std::thread bg([&] {
		f.cache.flush_to_disk(
			[&](bitfield&, span<disk_job* const>) -> int {
				std::unique_lock<std::mutex> lk(mtx);
				bg_parked = true;
				cv.notify_all();
				cv.wait(lk, [&] { return release_bg; });
				return 0;
			},
			0, // target=0, optimistic=false -> expensive pass flushes piece 0
			[](jobqueue_t, disk_job*) {},
			false);
	});

	// wait for the background flush to set flushing_flag on piece 0
	{
		std::unique_lock<std::mutex> lk(mtx);
		cv.wait(lk, [&] { return bg_parked; });
	}

	// two concurrent flush_storage() calls for the same storage
	auto storage_flusher = [&] {
		f.flush_storage_for();
		std::unique_lock<std::mutex> lk(mtx);
		++finished;
		cv.notify_all();
	};
	std::thread a(storage_flusher);
	std::thread b(storage_flusher);

	// wait until the skipper returns. A timeout guards against a regression
	// where neither skips and both wait forever on the background flush.
	{
		std::unique_lock<std::mutex> lk(mtx);
		bool const one_done =
			cv.wait_for(lk, std::chrono::seconds(10), [&] { return finished >= 1; });
		TEST_CHECK(one_done);
	}

	// release the background flush; the parked waiter now flushes and erases
	// piece 0.
	{
		std::unique_lock<std::mutex> lk(mtx);
		release_bg = true;
		cv.notify_all();
	}

	bg.join();
	a.join();
	b.join();

	TEST_EQUAL(int(f.cache.size()), 0);
	TEST_EQUAL(f.alloc.live, 0);
}

namespace {

struct flush_test_case
{
	// {initial, expected} per piece
	aux::vector<std::pair<std::string, std::string>, piece_index_t> pieces;
	int target;    // target block count left in cache after flush
	bool optimistic; // optimistic flush
};

void run_flush_test(flush_test_case tc)
{
	TORRENT_ASSERT(!tc.pieces.empty());
	piece_index_t const num_pieces = tc.pieces.end_index();

	// Erase '|' from each piece's initial state, remembering its position.
	// All pieces must have equal block count.
	aux::vector<int, piece_index_t> cursor(num_pieces);

	int n = -1;
	for (piece_index_t const p : tc.pieces.range())
	{
		auto& s = tc.pieces[p].first;
		auto const bar = s.find('|');
		TORRENT_ASSERT(bar != std::string::npos);
		cursor[p] = int(bar);
		s.erase(bar, 1);
		TORRENT_ASSERT(n == -1 || n == int(s.size()));
		n = int(s.size());
	}

	cache_fixture f(n, test_mode::v1);

	// Insert blocks before the cursor (hashed region) before kicking.
	// '.' and '!' both represent write_jobs in the cache; '!' means the
	// callback will refuse to flush that block (simulating a partial flush).
	for (piece_index_t const p : tc.pieces.range())
		for (int i = 0; i < cursor[p]; ++i)
		{
			char const c = tc.pieces[p].first[size_t(i)];
			if (c != '.' && c != '!') continue;
			f.insert(p, i);
		}

	// Kick the hasher, advancing all pieces to their cursor.
	f.kick_hashers();

	// Insert blocks at or after the cursor (unhashed) after the kick.
	for (piece_index_t const p : tc.pieces.range())
		for (int i = cursor[p]; i < n; ++i)
		{
			char const c = tc.pieces[p].first[size_t(i)];
			if (c != '.' && c != '!') continue;
			f.insert(p, i);
		}

	// Run the flush. The callback stamps '#' on flushed blocks directly into
	// the initial string. '!' blocks are present in the cache but skipped,
	// simulating a callback that only partially flushes a piece.
	f.cache.flush_to_disk(
		[&](bitfield& flushed, span<disk_job* const> blocks) -> int {
			int count = 0;
			for (int i = 0; i < blocks.size(); ++i)
			{
				auto const* wj = blocks[i];
				if (!wj) continue;
				auto const& w = std::get<job::write>(wj->action);
				auto const blk = static_cast<std::size_t>(w.offset / default_block_size);
				if (tc.pieces[w.piece].first[blk] == '!') continue;
				tc.pieces[w.piece].first[blk] = '#';
				flushed.set_bit(i);
				++count;
			}
			return count;
		},
		tc.target,
		[](jobqueue_t, disk_job*) {},
		tc.optimistic);

	for (piece_index_t const p : tc.pieces.range())
	{
		// '!' blocks were skipped by the callback; they remain in the cache
		// as ordinary unflushed blocks, indistinguishable from '.' in the result.
		for (char& c : tc.pieces[p].first)
			if (c == '!') c = '.';
		tc.pieces[p].first.insert(size_t(cursor[p]), 1, '|');
		TEST_EQUAL(tc.pieces[p].first, tc.pieces[p].second);
	}

	// Verify cache size and live buffer count: only '.' blocks (unflushed
	// write_jobs) hold a buffer. Flushed ('#') blocks have had their buffer
	// freed via free_disk_buffer(); the live count confirms this.
	int expected_cache_size = 0;
	for (auto const& [ini, exp] : tc.pieces)
		for (char c : exp)
			if (c == '.') ++expected_cache_size;
	TEST_EQUAL(int(f.cache.size()), expected_cache_size);
	TEST_EQUAL(f.alloc.live, expected_cache_size);
}

}

// State string encoding — one character per block, with a cursor marker:
//
//   '|'  marks the hasher cursor. Blocks to its left have been hashed;
//        blocks to its right have not.
//   '.'  block in cache
//   '!'  block in cache; flush callback will refuse to flush it (initial only)
//   '#'  flushed to disk (expected strings only)
//   ' '  block not in cache
flush_test_case const flush_cases[] = {

	// Optimistic flush only flushes blocks that are fully downloaded and fully
	// hashed, never causing disk read-backs or partial writes.
	{{
		{"...|", "###|"},
		{"..|.", "..|."}
	 }, 0, true},

	{{
		{"|.. ", "|.. "},
		{"..| ", "..| "},
		{"| ..", "| .."},
		{"...|", "###|"},
		{"|   ", "|   "},
		{"  |.", "  |."}
	 }, 0, true},

	// non-optimistic flush will force flush pieces even that haven not been
	// hashed yet, but prefer the ones that have been hashed first. pieces with
	// more cheap flushable blocks are flushed first. To avoid small writes, if
	// we start flushing a piece, flush as many hashed blocks as possible.
	{{
		{"..|..", "..|.."},
		{"...|.", "###|."}
	}, 6, false},
	{{
		{"..|..", "..|.."},
		{"...|.", "###|."}
	}, 5, false},
	{{
		{"..|..", "##|.."},
		{"...|.", "###|."}
	}, 4, false},
	{{
		{"..|..", "##|.."},
		{"...|.", "###|."}
	}, 3, false},
	{{
		{"..|..", "##|##"},
		{"...|.", "###|."}
	 }, 2, false},
	{{
		{"..|..", "##|##"},
		{"...|.", "###|."}
	 }, 1, false},
	{{
		{"..|..", "##|##"},
		{"...|.", "###|#"}
	 }, 0, false},

	// flushed_cursor only advances through leading contiguous flushed blocks.

	{{
		{".!.|", "#.#|"}
	 }, 0, false},

	// target=2 prevents phase 3 from flushing the remaining blocks.
	{{
		{".!.|.", "#.#|."}
	 }, 2, false},

	{{
		{"|.!.", "|#.#"}
	 }, 0, false},
};

TORRENT_TEST(flush_ordering)
{
	for (auto const& tc : flush_cases)
		run_flush_test(tc);
}

namespace {

// Tests a piece where piece_size2 < piece_size: a v2 file boundary falls
// inside what would be a full-sized v1 piece, truncating the v2 view.
//
// v2-only: the piece has 1 block of 11 bytes (piece_size == piece_size2 == 11).
// hybrid:  the piece has 2 full v1 blocks but piece_size2 == 11, so the v2
//          hasher only covers 11 bytes even though 2 * default_block_size bytes
//          are written.
void test_piece_size2_smaller_than_piece_size(test_mode_t const mode)
{
	constexpr int v1_piece_size = 2 * default_block_size;
	constexpr int piece_size2   = 11;

	bool const need_v1 = bool(mode & test_mode::v1);
	bool const need_v2 = bool(mode & test_mode::v2);

	// The effective piece size governs block layout: v1 size when v1 is
	// active, v2 size (piece_size2) for v2-only pieces.
	int const effective_piece_size = need_v1 ? v1_piece_size : piece_size2;
	int const blocks_in_piece = (effective_piece_size + default_block_size - 1)
		/ default_block_size;

	cache_fixture f(blocks_in_piece, mode);

	disk_cache::piece_entry_params const params{
		piece_size2, effective_piece_size, need_v1, need_v2, {}};

	for (int blk = 0; blk < blocks_in_piece; ++blk)
	{
		int const buf_size = std::min(default_block_size,
			effective_piece_size - blk * default_block_size);
		f.insert(0_piece, blk, params, buf_size);
	}

	// kick_hashers() also drains the v2 hash queue, populating f.v2_hashes.
	f.kick_hashers();

	int const v2_blocks = (piece_size2 + default_block_size - 1) / default_block_size;
	std::vector<sha256_hash> block_hashes(need_v2 ? v2_blocks : 0);
	auto hash_job = std::make_unique<pread_disk_job>();
	hash_job->action = job::hash{
		{}, 0_piece,
		span<sha256_hash>{block_hashes.data(), int(block_hashes.size())},
		sha1_hash{}
	};
	auto const hpr = f.cache.try_hash_piece(f.loc(0_piece), hash_job.get());
	if (need_v1)
	{
		TEST_EQUAL(hpr, disk_cache::hash_result::job_completed);
		TEST_CHECK(!std::get<job::hash>(hash_job->action).piece_hash.is_all_zeros());
	}
	else
	{
		// Pure-v2: try_hash_piece returns post_job (see comment in
		// test_disk_bottleneck).
		TEST_EQUAL(hpr, disk_cache::hash_result::post_job);
	}
	if (need_v2)
	{
		// v2 hashes were stashed via drain_v2_hash_queue into f.v2_hashes.
		TEST_EQUAL(int(f.v2_hashes.size()), v2_blocks);
		for (int b = 0; b < v2_blocks; ++b)
			TEST_CHECK(!f.v2_hashes.at({0_piece, b}).is_all_zeros());
	}

	TEST_EQUAL(f.flush(), blocks_in_piece);
	TEST_EQUAL(int(f.cache.size()), 0);
}

}

TORRENT_TEST(truncated_v2_piece_v2)
	{ test_piece_size2_smaller_than_piece_size(test_mode::v2); }
TORRENT_TEST(truncated_v2_piece_hybrid)
	{ test_piece_size2_smaller_than_piece_size(test_mode::v1 | test_mode::v2); }