A log scan whose end lies inside a disk-resident record returns that record truncated

# ScanEndRepro

Deterministic repro: a log scan whose end address falls inside a disk-resident record returns that
record truncated, and `CompactionType.Lookup` compaction persists the truncated copy while dropping
the original. Depends on `Tsavorite.core` only. Single thread, no checkpoints, no concurrent writer;
runs in a few seconds.

## Placement

Drop this folder into the Garnet repository as `playground/ScanEndRepro/`. The project file
references `..\..\libs\storage\Tsavorite\cs\src\core\Tsavorite.core.csproj` and pins `net10.0`.

## Running

```
dotnet run --project playground/ScanEndRepro -c Release -- <workdir>
```

Exit codes: `1` reproduced (damage with an end inside a record, none with a page-aligned end),
`0` not reproduced, `2` unexpected (damage also with a page-aligned end).

## What it does

1. Upserts 2,000 records with 32-byte keys and 2,000-byte values (`PageSize` 64 KB,
   `LogMemorySize` 256 KB), then `Log.FlushAndEvict(wait: true)` — the whole log is disk-resident.
2. Locates record 1000 with a scan to the tail and sets `end = address + 64`.
3. `Log.Scan(BeginAddress, end)`: prints the last record returned and checks its value.
4. `Compact(end, CompactionType.Lookup)`, then reads every key and checks every value.
5. Repeats 1–4 in a fresh store with `end` rounded down to a page boundary (control).

Every value starts with its id and continues with a non-zero pattern, so a value that was cut off
shows up as a run of zero bytes.

<Project Sdk="Microsoft.NET.Sdk">

	<PropertyGroup>
		<OutputType>Exe</OutputType>
		<!-- Single TFM: the repro uses `allows ref struct`, so net8.0 from the repo-wide
		     Directory.Build.props is not applicable here. -->
		<TargetFrameworks></TargetFrameworks>
		<TargetFramework>net10.0</TargetFramework>
		<Platforms>x64</Platforms>
		<AssemblyName>ScanEndRepro</AssemblyName>
		<IsPackable>false</IsPackable>
		<Nullable>enable</Nullable>
		<ImplicitUsings>enable</ImplicitUsings>
		<!-- Report numbers the same way on every machine, so pasted output is unambiguous. -->
		<InvariantGlobalization>true</InvariantGlobalization>
		<!-- Repro code, not shipping code: don't inherit the repo's strict gates. -->
		<TreatWarningsAsErrors>false</TreatWarningsAsErrors>
		<GenerateDocumentationFile>false</GenerateDocumentationFile>
		<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
		<SignAssembly>false</SignAssembly>
	</PropertyGroup>

	<ItemGroup>
		<!-- Path assumes this folder sits at <garnet>/playground/ScanEndRepro/. -->
		<ProjectReference Include="..\..\libs\storage\Tsavorite\cs\src\core\Tsavorite.core.csproj" />
	</ItemGroup>

</Project>

// Deterministic repro: a log scan whose end address falls inside a disk-resident record returns that
// record truncated, and Lookup compaction persists the truncated copy while dropping the original.
//
// Single thread, no checkpoints, no concurrent writer, a few seconds:
//
//   1. Upsert 2,000 records with ~2 KB values, then FlushAndEvict so the whole log is disk-resident.
//   2. Pick a record in the middle of the log; set the end address 64 bytes past its start.
//   3. Log.Scan(Begin, end): the last record returned is that record — with its value cut off at the
//      sector boundary that follows `end` (the frame is only loaded up to `end`, rounded up to a sector).
//   4. Compact(end, Lookup): the truncated record is copied to the tail, BeginAddress moves past the
//      original, and reading the key afterwards returns the damaged value.
//   5. Control: the same steps with `end` rounded down to a page boundary. No record straddles a page
//      boundary, so nothing is truncated and nothing is lost.
//
// Usage: ScanEndRepro [workdir]     Exit 1 = reproduced (damage with a mid-record end, none with a
// page-aligned end), 0 = not reproduced, 2 = unexpected (damage also with a page-aligned end).

using System.Runtime.InteropServices;
using Tsavorite.core;

using SF = Tsavorite.core.StoreFunctions<Key32Comparer, Tsavorite.core.DefaultRecordTriggers>;
using A = Tsavorite.core.SpanByteAllocator<Tsavorite.core.StoreFunctions<Key32Comparer, Tsavorite.core.DefaultRecordTriggers>>;

const long PageSize = 1L << 16;        // 64 KB
const long LogMemorySize = 1L << 18;   // 256 KB — the log below is many times larger
const int Records = 2000;
const int ValueSize = 2000;
const ulong VictimId = Records / 2;

var root = args.Length > 0 ? args[0] : Path.Combine(Path.GetTempPath(), "tsavorite-scan-end-repro");

var damagedMidRecord = Run(Path.Combine(root, "end-inside-record"), alignToPage: false);
var damagedAligned = Run(Path.Combine(root, "end-on-page-boundary"), alignToPage: true);

Console.WriteLine();
Console.WriteLine(
	$"SUMMARY: compaction end inside a record: {damagedMidRecord} damaged record(s); " +
	$"compaction end on a page boundary: {damagedAligned} damaged record(s).");

return damagedAligned > 0 ? 2 : damagedMidRecord > 0 ? 1 : 0;

static int Run(string dir, bool alignToPage)
{
	if (Directory.Exists(dir))
	{
		Directory.Delete(dir, recursive: true);
	}

	Console.WriteLine();
	Console.WriteLine($"== {(alignToPage ? "End rounded down to a page boundary (control)" : "End 64 bytes inside a record")}");

	using var device = Devices.CreateLogDevice(Path.Combine(dir, "hlog.log"));
	using var store = new TsavoriteKV<SF, A>(
		new KVSettings
		{
			IndexSize = 1L << 20,
			LogMemorySize = LogMemorySize,
			PageSize = PageSize,
			LogDevice = device,
			CheckpointDir = dir,
		},
		StoreFunctions.Create(Key32Comparer.Instance, DefaultRecordTriggers.Instance),
		(allocatorSettings, storeFunctions) => new(allocatorSettings, storeFunctions));

	using var session = store.NewSession<Key32, PinnedSpanByte, byte[]?, Empty, ReproFunctions>(ReproFunctions.Instance);
	var context = session.BasicContext;

	// 1. Write, then make everything disk-resident.
	var value = new byte[ValueSize];

	for (ulong id = 1; id <= Records; id++)
	{
		Fill(value, id);

		var status = context.Upsert(new Key32(id), value);

		if (status.IsFaulted)
		{
			throw new InvalidOperationException($"Upsert {id}: {status}");
		}

		_ = context.CompletePending(wait: true);
	}

	store.Log.FlushAndEvict(wait: true);

	Console.WriteLine(
		$"   log: Begin {store.Log.BeginAddress}, Head {store.Log.HeadAddress}, " +
		$"SafeReadOnly {store.Log.SafeReadOnlyAddress}, Tail {store.Log.TailAddress}");

	// 2. Locate the victim record (a scan to the tail never ends inside a record).
	long victimAddress = 0, victimSize = 0;

	using (var iterator = store.Log.Scan(store.Log.BeginAddress, store.Log.TailAddress))
	{
		while (iterator.GetNext())
		{
			if (MemoryMarshal.Read<ulong>(iterator.Key) == VictimId)
			{
				victimAddress = iterator.CurrentAddress;
				victimSize = iterator.NextAddress - iterator.CurrentAddress;
				break;
			}
		}
	}

	var end = victimAddress + 64;

	if (alignToPage)
	{
		end -= end % PageSize;
	}

	Console.WriteLine($"   victim record id {VictimId}: address {victimAddress}, size {victimSize}; scan/compaction end {end}");

	// 3. The scan alone: what does the last record returned look like?
	using (var iterator = store.Log.Scan(store.Log.BeginAddress, end))
	{
		long lastAddress = 0;
		ulong lastId = 0;
		byte[]? lastValue = null;

		while (iterator.GetNext())
		{
			lastAddress = iterator.CurrentAddress;
			lastId = MemoryMarshal.Read<ulong>(iterator.Key);
			lastValue = iterator.ValueSpan.ToArray();
		}

		Console.WriteLine(
			$"   Scan(Begin, {end}) — last record returned: id {lastId} at {lastAddress}, " +
			Describe(lastValue, lastId));
	}

	// 4. Compact up to the same end, then read every key.
	_ = session.Compact(end, CompactionType.Lookup);

	Console.WriteLine($"   after Compact({end}, Lookup): Begin {store.Log.BeginAddress}, Tail {store.Log.TailAddress}");

	var damaged = 0;

	for (ulong id = 1; id <= Records; id++)
	{
		byte[]? output = null;

		var status = context.Read(new Key32(id), ref output);

		if (status.IsPending)
		{
			_ = context.CompletePendingWithOutputs(out var completed, wait: true);

			while (completed.Next())
			{
				status = completed.Current.Status;
				output = completed.Current.Output;
			}

			completed.Dispose();
		}

		var problem = status.Found ? Check(output!, id) : "NOT FOUND";

		if (problem is not null)
		{
			damaged++;
			Console.WriteLine($"   read id {id}: {problem}");
		}
	}

	Console.WriteLine($"   {damaged} of {Records} records damaged or missing");

	return damaged;
}

static string Describe(byte[]? value, ulong id)
	=> value is null ? "no record" : $"value length {value.Length}: {Check(value, id) ?? "intact"}";

// Value layout: bytes 0..7 = id, then a non-zero pattern derived from id and offset.
static void Fill(byte[] value, ulong id)
{
	MemoryMarshal.Write(value, in id);

	for (var i = 8; i < value.Length; i++)
	{
		value[i] = Pattern(id, i);
	}
}

static byte Pattern(ulong id, int offset) => (byte)(1 + (int)((id + (ulong)offset) % 251));

static string? Check(byte[] value, ulong id)
{
	if (value.Length != ValueSize)
	{
		return $"length {value.Length}, expected {ValueSize}";
	}

	if (MemoryMarshal.Read<ulong>(value) != id)
	{
		return $"id bytes wrong ({MemoryMarshal.Read<ulong>(value):x16})";
	}

	for (var i = 8; i < value.Length; i++)
	{
		if (value[i] != Pattern(id, i))
		{
			var zeroTail = value.AsSpan(i).IndexOfAnyExcept((byte)0) < 0;

			return $"TRUNCATED — value bytes from offset {i} on are {(zeroTail ? "all zero" : "wrong")}";
		}
	}

	return null;
}

// ============================== Tsavorite binding ==============================

/// <summary>32-byte key: 4 x 64 bit, expanded deterministically from a single id.</summary>
internal struct Key32 : IKey
{
	private ulong _a, _b, _c, _d;

	public Key32(ulong id)
	{
		_a = id;
		_b = id * 0x9E3779B97F4A7C15UL;
		_c = id ^ 0xC2B2AE3D27D4EB4FUL;
		_d = ~id;
	}

	public readonly bool IsPinned => false;

	public ReadOnlySpan<byte> KeyBytes
		=> MemoryMarshal.AsBytes(MemoryMarshal.CreateReadOnlySpan(ref _a, 4));

	public readonly bool HasNamespace => false;

	public readonly ReadOnlySpan<byte> NamespaceBytes => [];
}

internal struct Key32Comparer : IKeyComparer
{
	public static readonly Key32Comparer Instance = new();

	public readonly long GetHashCode64<TKey>(TKey key)
		where TKey : IKey, allows ref struct
		=> (long)MemoryMarshal.Read<ulong>(key.KeyBytes);

	public readonly bool Equals<TFirstKey, TSecondKey>(TFirstKey k1, TSecondKey k2)
		where TFirstKey : IKey, allows ref struct
		where TSecondKey : IKey, allows ref struct
		=> k1.KeyBytes.SequenceEqual(k2.KeyBytes);
}

/// <summary>Reader materializes the value span as an array.</summary>
internal sealed class ReproFunctions : SessionFunctionsBase<PinnedSpanByte, byte[]?, Empty>
{
	public static readonly ReproFunctions Instance = new();

	public override bool Reader<TSourceLogRecord>(in TSourceLogRecord srcLogRecord, ref PinnedSpanByte input, ref byte[]? output, ref ReadInfo readInfo)
	{
		output = srcLogRecord.ValueSpan.ToArray();

		return true;
	}
}

# TsavoriteRepro

Minimal repro for silent data loss during `CompactionType.Lookup` over a disk-resident log
region. Depends on `Tsavorite.core` only.

## Placement

Drop this folder into the Garnet repository as `playground/TsavoriteRepro/`. The project file
references `..\..\libs\storage\Tsavorite\cs\src\core\Tsavorite.core.csproj` and pins `net10.0`
(the code uses `allows ref struct`, so the repo-wide `net8.0` target does not apply).

## Running

```
# damage condition — expected: reproduces, exit 1
dotnet run --project playground/TsavoriteRepro -c Release -- 3 <workdir>

# control run — expected: clean across all iterations, exit 0
dotnet run --project playground/TsavoriteRepro -c Release -- 10 <workdir> safe

# same physics as unsafe, every compaction end rounded down to a page boundary — expected: clean, exit 0
dotnet run --project playground/TsavoriteRepro -c Release -- 10 <workdir> aligned
```

Arguments: iterations (default 3), working directory (default `%TEMP%`), mode
(`unsafe` default | `safe` | `aligned`). All iterations always run — the damage is timing dependent, so the
run reports a hit rate rather than stopping at the first hit; directories of damaged iterations
are left behind for inspection. Roughly 70–130 s per iteration; each iteration needs a few GB of
free disk space.

Exit codes: `0` clean, `1` data loss in at least one iteration, `2` no data loss but `Compact()`
threw (compaction barely ran, so the run proves nothing).

## What it does

A single writer fills the store with 500,000 write-once "cold" records and 1,000 "hub" records
that grow by 32 B per read-modify-write, so every previous hub version becomes garbage. A
background thread compacts in nibbles, another takes FoldOver checkpoints every 50 ms; the two
are strictly serialized against each other, so no compaction ever overlaps a checkpoint.

The decisive setting is the compaction threshold (160 MB) being **above** `LogMemorySize`
(128 MB): compaction only starts once the backlog no longer fits in memory, so every nibble
works below `HeadAddress` and both the liveness probe and copy-to-tail take the pending/disk
path.

Afterwards every expected key is audited (length plus a content probe) twice: once against the
live store, once after a full checkpoint, dispose, and recovery into a fresh store.

In `safe` mode the threshold is small and compaction additionally only runs while the
untruncated log fits into the memory window minus a margin, so every nibble stays above
`HeadAddress`. That run stays clean — it is the control that isolates the disk-resident path as
the trigger.

In `aligned` mode the physics are those of `unsafe`, but every compaction end address is rounded
down to a page boundary. That run stays clean too: the damage comes from the record that straddles
the compaction end, which a page-aligned end never cuts (see `../scan-end-repro` for the
deterministic form of the same defect).

<Project Sdk="Microsoft.NET.Sdk">

	<PropertyGroup>
		<OutputType>Exe</OutputType>
		<!-- Single TFM: the repro uses `allows ref struct`, so net8.0 from the repo-wide
		     Directory.Build.props is not applicable here. -->
		<TargetFrameworks></TargetFrameworks>
		<TargetFramework>net10.0</TargetFramework>
		<Platforms>x64</Platforms>
		<AssemblyName>TsavoriteRepro</AssemblyName>
		<IsPackable>false</IsPackable>
		<Nullable>enable</Nullable>
		<ImplicitUsings>enable</ImplicitUsings>
		<!-- Report numbers the same way on every machine, so pasted output is unambiguous. -->
		<InvariantGlobalization>true</InvariantGlobalization>
		<!-- Repro code, not shipping code: don't inherit the repo's strict gates. -->
		<TreatWarningsAsErrors>false</TreatWarningsAsErrors>
		<GenerateDocumentationFile>false</GenerateDocumentationFile>
		<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
		<SignAssembly>false</SignAssembly>
	</PropertyGroup>

	<ItemGroup>
		<!-- Path assumes this folder sits at <garnet>/playground/TsavoriteRepro/. -->
		<ProjectReference Include="..\..\libs\storage\Tsavorite\cs\src\core\Tsavorite.core.csproj" />
	</ItemGroup>

</Project>

// Minimal repro: Lookup compaction over a disk-resident log region loses records.
//
// Depends on Tsavorite.core only. Observed damage: records with the correct length but an
// all-zero value payload, and records that are gone entirely. Both are visible in the live
// store (before any recovery) and survive checkpoint + recovery.
//
// Physics:
//
//   - 32-byte keys (hash = leading 8 bytes), values are byte blobs
//   - "hubs": values grown by 32 B per read-modify-write, so every previous version is garbage
//   - "cold": write-once keys that age below HeadAddress
//   - LogMemorySize 128 MB, PageSize 32 MB, FoldOver checkpoints every 50 ms
//   - ONE writer thread; compaction nibbles (Lookup) and checkpoints run on their own threads
//     and are strictly SERIALIZED against each other by a semaphore — overlap is demonstrably
//     not required for the damage
//   - THE key ingredient: the compaction threshold is set ABOVE the log memory size, so every
//     nibble works below HeadAddress and both the liveness probe and copy-to-tail go through
//     the pending/disk path. If compaction keeps up instead (small backlog, memory-resident),
//     the damage does NOT occur — see the "safe" mode below.
//
// Verification: full audit of every expected key (length + content probe), once LIVE after the
// run and once after a full checkpoint + dispose + recovery into a fresh store.
// Exit 1 on loss/staleness; the directory is then left behind for inspection.
//
// Usage: TsavoriteRepro [iterations] [rootDirectory] [unsafe|safe|aligned]   (defaults: 3, %TEMP%, unsafe)

using System.Diagnostics;
using System.Runtime.InteropServices;
using Tsavorite.core;

using SF = Tsavorite.core.StoreFunctions<Key32Comparer, Tsavorite.core.DefaultRecordTriggers>;
using A = Tsavorite.core.SpanByteAllocator<Tsavorite.core.StoreFunctions<Key32Comparer, Tsavorite.core.DefaultRecordTriggers>>;

// ============================== Configuration ==============================

const long LogMemorySize = 128L * 1024 * 1024;
const long PageSize = 32L * 1024 * 1024;

const int ColdKeys = 500_000;      // written once, age out to disk
const int Hubs = 1000;             // grow by 32 B per rewrite (read-modify-write)
const int HubRewriteEvery = 3;     // every 3rd operation is a hub rewrite

// Interleaving: the writer is throttled so that many nibbles and checkpoints run WHILE writing
// is in progress, and live cold records travel across several copy generations.
const int WriterPauseEveryOps = 1_000;
const int WriterPauseMs = 2;
const int CompactionIntervalMs = 10;
const int CheckpointIntervalMs = 50;

var iterations = args.Length > 0 ? int.Parse(args[0]) : 3;
var root = args.Length > 1 ? args[1] : Path.Combine(Path.GetTempPath(), "tsavorite-compaction-repro");

// Mode (args[2]):
//   "unsafe" (default) — threshold DELIBERATELY above the log memory size: compaction only
//   starts once the backlog exceeds the in-memory window, so every nibble works below
//   HeadAddress and goes through the pending/disk path. Expectation: REPRODUCES (exit 1).
//   "safe" — control run: small threshold (compaction keeps up) PLUS the guard condition
//   "untruncated log <= memory window - margin", so every nibble stays above HeadAddress.
//   Expectation: CLEAN across all iterations (exit 0). This is the boundary that separates
//   damaged from undamaged runs in our measurements.
var safeMode = args.Length > 2 && args[2] == "safe";
// "aligned": unsafe physics, but every compaction end address is rounded down to a page boundary, so no record straddles the scan end.
var aligned = args.Length > 2 && args[2] == "aligned";

var compactionThreshold = safeMode ? 16L * 1024 * 1024 : 160L * 1024 * 1024;
var compactionStep = safeMode ? 16L * 1024 * 1024 : 128L * 1024 * 1024;
var guardMargin = safeMode ? 48L * 1024 * 1024 : 0L; // > 1 page (32 MB) + writer burst

Console.WriteLine($"Mode: {(aligned ? "aligned (unsafe physics, page-aligned compaction end)" : safeMode ? "safe (control run, expects 0 hits)" : "unsafe (damage condition, expects repro)")}");

var totalCompactionFailures = 0;
var damagedIterations = 0;

for (var iteration = 1; iteration <= iterations; iteration++)
{
	var dir = Path.Combine(root, $"iter{iteration}");

	if (Directory.Exists(dir))
	{
		Directory.Delete(dir, recursive: true);
	}

	Console.WriteLine($"== Iteration {iteration} (directory {dir})");

	var (liveBad, recoveredBad, threwDuringCompaction) = RunIteration(dir, seed: iteration * 7919, compactionThreshold, compactionStep, guardMargin, aligned);

	Console.WriteLine(
		$"== Iteration {iteration}: {liveBad} errors live, {recoveredBad} errors after recovery, " +
		$"{threwDuringCompaction} Compact() failures");

	totalCompactionFailures += threwDuringCompaction;

	// Deliberately keep going instead of stopping at the first hit: the damage is timing
	// dependent, so a hit rate over all iterations says more than a single anecdote.
	if (liveBad + recoveredBad > 0)
	{
		damagedIterations++;

		Console.WriteLine($"REPRODUCED — directory left behind: {dir}");

		continue;
	}

	Directory.Delete(dir, recursive: true);
}

Console.WriteLine(
	$"SUMMARY: {damagedIterations} of {iterations} iterations lost data; " +
	$"Compact() threw {totalCompactionFailures} times in total.");

if (damagedIterations == 0 && totalCompactionFailures > 0)
{
	Console.WriteLine(
		"No data loss, but compaction kept throwing -- it barely ran, so this is NOT a clean result.");
}

return damagedIterations > 0 ? 1 : totalCompactionFailures > 0 ? 2 : 0;

// ============================== One iteration ==============================

static (int liveBad, int recoveredBad, int compactionFailures) RunIteration(string dir, int seed, long compactionThreshold, long compactionStep, long guardMargin, bool aligned)
{
	var stopwatch = Stopwatch.StartNew();

	// Expectations: hub h holds hubSizes[h] bytes after the run; cold i holds coldSizes[i] bytes.
	var hubSizes = new int[Hubs];
	var coldSizes = new int[ColdKeys];

	long liveBad, recoveredBad = 0;

	var compactionFailures = 0;
	var checkpointFailures = 0;

	using (var log = Devices.CreateLogDevice(Path.Combine(dir, "Log")))
	{
		using var store = CreateStore(log, dir);

		using var writerSession = store.NewSession<Key32, PinnedSpanByte, byte[]?, Empty, ReproFunctions>(ReproFunctions.Instance);
		using var compactionSession = store.NewSession<Key32, PinnedSpanByte, byte[]?, Empty, ReproFunctions>(ReproFunctions.Instance);

		// Checkpoint <-> compaction strictly serialized.
		using var exclusion = new SemaphoreSlim(1, 1);
		using var stop = new CancellationTokenSource();

		var compactionRounds = 0;

		// Threshold in a box so the drain phase can lower it (otherwise the last backlog would
		// stay behind and the drain would spin without effect).
		var threshold = new[] { compactionThreshold };

		var compactor = Task.Run(() =>
		{
			while (!stop.IsCancellationRequested)
			{
				Thread.Sleep(CompactionIntervalMs);

				exclusion.Wait();

				try
				{
					var l = store.Log;

					// Safe mode: only compact while the untruncated log fits into the memory
					// window with margin, keeping the region safely above HeadAddress.
					if (guardMargin > 0 && l.TailAddress - l.BeginAddress > LogMemorySize - guardMargin)
					{
						continue;
					}

					if (l.SafeReadOnlyAddress - l.BeginAddress > Volatile.Read(ref threshold[0]))
					{
						var until = Math.Min(l.BeginAddress + compactionStep, l.SafeReadOnlyAddress);
						if (aligned)
						{
							until &= ~(PageSize - 1);
							if (until <= l.BeginAddress)
								continue;
						}

						_ = compactionSession.Compact(until, CompactionType.Lookup);

						Interlocked.Increment(ref compactionRounds);
					}
				}
				catch (Exception exception)
				{
					// An exception escaping Compact() leaves the store in an undefined state:
					// part of the region may already have been copied to the tail. Print the
					// first one in full -- the stack trace is the interesting artifact.
					if (Interlocked.Increment(ref compactionFailures) == 1)
					{
						Console.WriteLine($"   Compact() threw (first occurrence):{Environment.NewLine}{exception}");
					}
				}
				finally
				{
					_ = exclusion.Release();
				}
			}
		});

		var checkpointCount = 0;

		var checkpointer = Task.Run(() =>
		{
			while (!stop.IsCancellationRequested)
			{
				Thread.Sleep(CheckpointIntervalMs);

				exclusion.Wait();

				try
				{
					if (store.TryInitiateHybridLogCheckpoint(out _, CheckpointType.FoldOver))
					{
						store.CompleteCheckpointAsync().AsTask().Wait();

						Interlocked.Increment(ref checkpointCount);
					}
				}
				catch (Exception exception)
				{
					if (Interlocked.Increment(ref checkpointFailures) == 1)
					{
						Console.WriteLine($"   Checkpoint threw (first occurrence):{Environment.NewLine}{exception}");
					}
				}
				finally
				{
					_ = exclusion.Release();
				}
			}
		});

		// ---------------- Writer (single) ----------------

		var context = writerSession.BasicContext;
		var buffer = new byte[Hubs * 32 * 4096 / Hubs]; // generous; trimmed per operation
		var rng = new Random(seed);

		var totalOps = ColdKeys + ColdKeys / HubRewriteEvery;
		var cold = 0;

		for (var op = 0; op < totalOps; op++)
		{
			if (op % WriterPauseEveryOps == 0)
			{
				// Throttle so that nibbles and checkpoints run heavily WHILE writing.
				Thread.Sleep(WriterPauseMs);
			}

			if (op % (HubRewriteEvery + 1) == HubRewriteEvery)
			{
				// Hub rewrite as read-modify-write: read first (exercising the pending read path
				// concurrently with compaction), then write the grown value back.
				var h = rng.Next(Hubs);

				byte[]? current = null;
				var read = context.Read(HubKey(h), ref current);

				if (read.IsPending)
				{
					_ = Resolve(writerSession, ref current);
				}

				hubSizes[h] += 32;

				var span = buffer.AsSpan(0, hubSizes[h]);

				Fill(span, HubKeyId(h), hubSizes[h]);

				Upsert(context, HubKey(h), span);
			}
			else
			{
				// Cold: written once, never touched again.
				var size = 64 + rng.Next(448);

				coldSizes[cold] = size;

				var span = buffer.AsSpan(0, size);

				Fill(span, ColdKeyId(cold), size);

				Upsert(context, ColdKey(cold), span);

				cold++;
			}
		}

		// Drain: compact the remaining cold backlog away while checkpoints keep running.
		Volatile.Write(ref threshold[0], 16L * 1024 * 1024);

		var drainDeadline = DateTime.UtcNow.AddSeconds(60);

		while (store.Log.SafeReadOnlyAddress - store.Log.BeginAddress > (aligned ? 64L : 24L) * 1024 * 1024
			&& DateTime.UtcNow < drainDeadline)
		{
			Thread.Sleep(50);
		}

		stop.Cancel();
		Task.WaitAll(compactor, checkpointer);

		Console.WriteLine(
			$"   Build-up: {totalOps:N0} ops ({cold:N0} cold, {totalOps - cold:N0} hub rewrites), " +
			$"log {(store.Log.TailAddress - store.Log.BeginAddress) / (1 << 20):N0} MB, " +
			$"{compactionRounds} nibbles, {checkpointCount} checkpoints, " +
			$"{compactionFailures} compaction failures, {checkpointFailures} checkpoint failures, " +
			$"{stopwatch.Elapsed.TotalSeconds:N0} s");

		// ---------------- Audit 1: the live view ----------------

		liveBad = Audit(writerSession, hubSizes, coldSizes, cold, "live");

		// ---------------- Full checkpoint + clean shutdown ----------------

		while (!store.TryInitiateFullCheckpoint(out _, CheckpointType.FoldOver))
		{
			store.CompleteCheckpointAsync().AsTask().Wait();
		}

		store.CompleteCheckpointAsync().AsTask().Wait();
	}

	// ---------------- Audit 2: after recovery into a fresh store ----------------

	using (var log = Devices.CreateLogDevice(Path.Combine(dir, "Log")))
	{
		using var store = CreateStore(log, dir);

		_ = store.RecoverAsync().AsTask().Result;

		using var session = store.NewSession<Key32, PinnedSpanByte, byte[]?, Empty, ReproFunctions>(ReproFunctions.Instance);

		recoveredBad = Audit(session, hubSizes, coldSizes, ColdKeys, "after recovery");
	}

	return ((int)liveBad, (int)recoveredBad, compactionFailures);
}

// ============================== Building blocks ==============================

static TsavoriteKV<SF, A> CreateStore(IDevice log, string dir)
	=> new(
		new KVSettings
		{
			IndexSize = 1L << 26,
			LogMemorySize = LogMemorySize,
			PageSize = PageSize,
			LogDevice = log,
			CheckpointDir = dir,
			RemoveOutdatedCheckpoints = true,
		},
		StoreFunctions.Create(Key32Comparer.Instance, DefaultRecordTriggers.Instance),
		(allocatorSettings, storeFunctions) => new(allocatorSettings, storeFunctions));

static void Upsert(
	BasicContext<Key32, PinnedSpanByte, byte[]?, Empty, ReproFunctions, SF, A> context,
	Key32 key, ReadOnlySpan<byte> value)
{
	var status = context.Upsert(key, value);

	if (status.IsFaulted)
	{
		throw new InvalidOperationException($"Upsert: {status}");
	}

	_ = context.CompletePending(wait: true);
}

static int Audit(
	ClientSession<Key32, PinnedSpanByte, byte[]?, Empty, ReproFunctions, SF, A> session,
	int[] hubSizes, int[] coldSizes, int coldCount, string stage)
{
	var bad = 0;
	var examples = new List<string>();

	void Check(Key32 key, ulong id, int expectedSize, string label)
	{
		byte[]? output = null;

		var status = session.BasicContext.Read(key, ref output);

		if (status.IsPending)
		{
			status = Resolve(session, ref output);
		}

		var problem = status.Found
			? Describe(output!, id, expectedSize)
			: "missing entirely (Read -> NOTFOUND)";

		if (problem is null)
		{
			return;
		}

		bad++;

		if (examples.Count < 12)
		{
			examples.Add($"      {label}: {problem}");
		}
	}

	for (var h = 0; h < hubSizes.Length; h++)
	{
		if (hubSizes[h] > 0)
		{
			Check(HubKey(h), HubKeyId(h), hubSizes[h], $"Hub {h}");
		}
	}

	for (var c = 0; c < coldCount; c++)
	{
		Check(ColdKey(c), ColdKeyId(c), coldSizes[c], $"Cold {c}");
	}

	Console.WriteLine($"   Audit ({stage}): {bad} errors");

	foreach (var example in examples)
	{
		Console.WriteLine(example);
	}

	return bad;
}

static string? Describe(byte[] actual, ulong id, int expectedSize)
{
	if (actual.Length != expectedSize)
	{
		return $"length {actual.Length} instead of {expectedSize} (stale version)";
	}

	Span<byte> expected = stackalloc byte[Math.Min(expectedSize, 16)];

	Fill(expected, id, expectedSize);

	if (actual.AsSpan(0, expected.Length).SequenceEqual(expected))
	{
		return null;
	}

	// Forensics: what is in there instead? By construction the first 8 bytes are the writer's
	// id and the next 8 are the size. Observed signature: length correct, payload all zero --
	// the copy was made from a scan image that was cut off at a sector boundary (see ../scan-end-repro).
	var foundId = actual.Length >= 8 ? MemoryMarshal.Read<ulong>(actual) : 0UL;
	var foundSize = actual.Length >= 16 ? MemoryMarshal.Read<ulong>(actual.AsSpan(8)) : 0UL;

	var kind = actual.All(b => b == 0)
		? "VALUE ENTIRELY ZEROED"
		: foundId == id ? "same key, different bytes" : "foreign content";

	return $"content mismatch: expected id={id:x16}/size={expectedSize}, " +
		$"found id={foundId:x16}/size={foundSize} ({kind})";
}

static Status Resolve(
	ClientSession<Key32, PinnedSpanByte, byte[]?, Empty, ReproFunctions, SF, A> session,
	ref byte[]? output)
{
	_ = session.BasicContext.CompletePendingWithOutputs(out var completed, wait: true);

	Status status = default;

	while (completed.Next())
	{
		status = completed.Current.Status;
		output = completed.Current.Output;
	}

	completed.Dispose();

	return status;
}

// Deterministic content: first 8 bytes = id, then the size, then zero padding.
static void Fill(Span<byte> span, ulong id, int size)
{
	span.Clear();

	if (span.Length >= 8)
	{
		MemoryMarshal.Write(span, in id);
	}

	if (span.Length >= 16)
	{
		var s = (ulong)size;
		MemoryMarshal.Write(span[8..], in s);
	}
}

static Key32 ColdKey(int index) => new(ColdKeyId(index));

static Key32 HubKey(int index) => new(HubKeyId(index));

static ulong ColdKeyId(int index) => SplitMix((ulong)index + 1);

static ulong HubKeyId(int index) => SplitMix(0x8000_0000_0000_0000UL + (ulong)index);

static ulong SplitMix(ulong x)
{
	x += 0x9E3779B97F4A7C15UL;
	x = (x ^ (x >> 30)) * 0xBF58476D1CE4E5B9UL;
	x = (x ^ (x >> 27)) * 0x94D049BB133111EBUL;

	return x ^ (x >> 31);
}

// ============================== Tsavorite binding ==============================

/// <summary>32-byte key: 4 x 64 bit, expanded deterministically from a single id.</summary>
internal struct Key32 : IKey
{
	private ulong _a, _b, _c, _d;

	public Key32(ulong id)
	{
		_a = id;
		_b = id * 0x9E3779B97F4A7C15UL;
		_c = id ^ 0xC2B2AE3D27D4EB4FUL;
		_d = ~id;
	}

	public readonly bool IsPinned => false;

	public ReadOnlySpan<byte> KeyBytes
		=> MemoryMarshal.AsBytes(MemoryMarshal.CreateReadOnlySpan(ref _a, 4));

	public readonly bool HasNamespace => false;

	public readonly ReadOnlySpan<byte> NamespaceBytes => [];
}

internal struct Key32Comparer : IKeyComparer
{
	public static readonly Key32Comparer Instance = new();

	public readonly long GetHashCode64<TKey>(TKey key)
		where TKey : IKey, allows ref struct
		=> (long)MemoryMarshal.Read<ulong>(key.KeyBytes);

	public readonly bool Equals<TFirstKey, TSecondKey>(TFirstKey k1, TSecondKey k2)
		where TFirstKey : IKey, allows ref struct
		where TSecondKey : IKey, allows ref struct
		=> k1.KeyBytes.SequenceEqual(k2.KeyBytes);
}

/// <summary>Reader materializes the value span as an array.</summary>
internal sealed class ReproFunctions : SessionFunctionsBase<PinnedSpanByte, byte[]?, Empty>
{
	public static readonly ReproFunctions Instance = new();

	public override bool Reader<TSourceLogRecord>(in TSourceLogRecord srcLogRecord, ref PinnedSpanByte input, ref byte[]? output, ref ReadInfo readInfo)
	{
		output = srcLogRecord.ValueSpan.ToArray();

		return true;
	}
}