laburnum 1.17.0

// Copyright Two Neutron Stars Incorporated and contributors
// SPDX-License-Identifier: BlueOak-1.0.0

//! Tests for the tri-color mark-sweep garbage collector.
//!
//! These tests exercise GarbageCollector in isolation — no scheduler.
//! Records are inserted directly into PartitionStore instances.

use {
  crate::{
    ContentHash,
    Ident,
    database::{
      ContentHashRef,
      Database,
      GenerationEpoch,
      HasPartition,
      PartitionKey,
      PartitionStore,
      chunk::RecordWriter,
      gc::{GarbageCollector, GcPhase},
      storage::Partitions,
    },
    record::Record,
  },
  super::storage::{
    TestPartition,
    Test1Partition,
    TestPartitions,
    TestRecordData,
  },
};

fn make_record(name: &str) -> TestRecordData {
  TestRecordData::Module {
    exports: vec![Ident::new(name)],
  }
}

fn make_record_with_refs(children: Vec<ContentHash>) -> TestRecordData {
  TestRecordData::WithRefs { children }
}

fn hash_of(record: &TestRecordData) -> ContentHash {
  record.content_hash()
}

/// Wrap a ContentHash as a ContentHashRef for TestPartition.
fn test_ref(hash: ContentHash) -> ContentHashRef {
  ContentHashRef::new(TestPartition::KEY, hash)
}

/// Wrap a ContentHash as a ContentHashRef for Test1Partition.
fn test1_ref(hash: ContentHash) -> ContentHashRef {
  ContentHashRef::new(Test1Partition::KEY, hash)
}

/// Insert a record into TestPartition and return its hash.
fn insert_test(
  stores: &<TestPartitions as Partitions>::Stores,
  record: TestRecordData,
  epoch: GenerationEpoch,
) -> ContentHash {
  let hash = hash_of(&record);
  let store: &PartitionStore<TestPartition> = stores.store();
  store.insert(hash, record, epoch);
  hash
}

/// Insert a record into Test1Partition and return its hash.
fn insert_test1(
  stores: &<TestPartitions as Partitions>::Stores,
  record: TestRecordData,
  epoch: GenerationEpoch,
) -> ContentHash {
  let hash = hash_of(&record);
  let store: &PartitionStore<Test1Partition> = stores.store();
  store.insert(hash, record, epoch);
  hash
}

fn store_contains(
  stores: &<TestPartitions as Partitions>::Stores,
  hash: &ContentHash,
) -> bool {
  let store: &PartitionStore<TestPartition> = stores.store();
  store.get(hash).is_some()
}

fn store_len(
  stores: &<TestPartitions as Partitions>::Stores,
) -> usize {
  let store: &PartitionStore<TestPartition> = stores.store();
  store.len()
}

fn store1_len(
  stores: &<TestPartitions as Partitions>::Stores,
) -> usize {
  let store: &PartitionStore<Test1Partition> = stores.store();
  store.len()
}

/// Run mark ticks on TestPartition only until gray queue is drained.
///
/// Use this for tests where all records are in TestPartition.
fn mark_to_completion(
  gc: &GarbageCollector,
  stores: &<TestPartitions as Partitions>::Stores,
) {
  for _ in 0..1000 {
    let done = gc.mark_tick_partition::<TestPartitions, TestPartition>(stores, 100);
    if done {
      return;
    }
  }
  panic!("marking did not complete after 1000 ticks");
}

#[test]
fn idle_to_marking_to_sweeping_to_idle() {
  let gc = GarbageCollector::new();
  assert_eq!(gc.phase(), GcPhase::Idle);
  assert!(!gc.is_active());
  assert!(!gc.is_marking());

  assert!(gc.start_marking(std::iter::empty()));
  assert_eq!(gc.phase(), GcPhase::Marking);
  assert!(gc.is_active());
  assert!(gc.is_marking());

  assert!(gc.finish_marking());
  assert_eq!(gc.phase(), GcPhase::Sweeping);
  assert!(gc.is_active());
  assert!(!gc.is_marking());

  gc.finish_sweep();
  assert_eq!(gc.phase(), GcPhase::Idle);
  assert!(!gc.is_active());
  assert!(!gc.is_marking());
}

#[test]
fn start_marking_rejects_when_already_marking() {
  let gc = GarbageCollector::new();
  assert!(gc.start_marking(std::iter::empty()));
  assert!(!gc.start_marking(std::iter::empty()));
  assert_eq!(gc.phase(), GcPhase::Marking);
}

#[test]
fn start_marking_rejects_when_sweeping() {
  let gc = GarbageCollector::new();
  assert!(gc.start_marking(std::iter::empty()));
  assert!(gc.finish_marking());
  assert_eq!(gc.phase(), GcPhase::Sweeping);
  assert!(!gc.start_marking(std::iter::empty()));
  assert_eq!(gc.phase(), GcPhase::Sweeping);
}

#[test]
fn finish_marking_fails_when_gray_not_empty() {
  let gc = GarbageCollector::new();
  let hash = ContentHash::new(&[1, 2, 3]);
  assert!(gc.start_marking(std::iter::once(test_ref(hash))));

  // Gray has one entry — finish_marking should fail
  assert!(!gc.finish_marking());
  assert_eq!(gc.phase(), GcPhase::Marking);
}

#[test]
fn finish_marking_succeeds_when_gray_empty() {
  let gc = GarbageCollector::new();
  assert!(gc.start_marking(std::iter::empty()));
  assert!(gc.finish_marking());
  assert_eq!(gc.phase(), GcPhase::Sweeping);
}

#[test]
fn epoch_increments_each_cycle() {
  let gc = GarbageCollector::new();
  let e0 = gc.epoch();

  // Cycle 1
  gc.start_marking(std::iter::empty());
  let e1 = gc.epoch();
  gc.finish_marking();
  gc.finish_sweep();

  // Cycle 2
  gc.start_marking(std::iter::empty());
  let e2 = gc.epoch();
  gc.finish_marking();
  gc.finish_sweep();

  assert_eq!(e1.get(), e0.get() + 1);
  assert_eq!(e2.get(), e0.get() + 2);
}

#[test]
fn single_root_no_references() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();

  let r1 = insert_test(&stores, make_record("r1"), GenerationEpoch::new(0));

  gc.start_marking(std::iter::once(test_ref(r1)));
  mark_to_completion(&gc, &stores);

  assert!(gc.is_black(&r1));
  assert_eq!(gc.black_set_len(), 1);
}

#[test]
fn single_root_with_direct_reference() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();

  let r2 = insert_test(&stores, make_record("r2"), GenerationEpoch::new(0));
  let r1 = insert_test(
    &stores,
    make_record_with_refs(vec![r2]),
    GenerationEpoch::new(0),
  );

  gc.start_marking(std::iter::once(test_ref(r1)));
  mark_to_completion(&gc, &stores);

  assert!(gc.is_black(&r1));
  assert!(gc.is_black(&r2));
}

#[test]
fn multiple_independent_roots() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();

  let r1 = insert_test(&stores, make_record("r1"), GenerationEpoch::new(0));
  let r2 = insert_test(&stores, make_record("r2"), GenerationEpoch::new(0));
  let r3 = insert_test(&stores, make_record("r3"), GenerationEpoch::new(0));

  gc.start_marking(vec![test_ref(r1), test_ref(r2), test_ref(r3)].into_iter());
  mark_to_completion(&gc, &stores);

  assert!(gc.is_black(&r1));
  assert!(gc.is_black(&r2));
  assert!(gc.is_black(&r3));
  assert_eq!(gc.black_set_len(), 3);
}

#[test]
fn unreferenced_record_not_marked() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();

  let root = insert_test(&stores, make_record("root"), GenerationEpoch::new(0));
  let orphan = insert_test(&stores, make_record("orphan"), GenerationEpoch::new(0));

  gc.start_marking(std::iter::once(test_ref(root)));
  mark_to_completion(&gc, &stores);

  assert!(gc.is_black(&root));
  assert!(!gc.is_black(&orphan));
}

#[test]
fn root_hash_not_in_store() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();

  let phantom_hash = ContentHash::new(&[99, 99, 99]);

  gc.start_marking(std::iter::once(test_ref(phantom_hash)));
  mark_to_completion(&gc, &stores);

  // Goes to black even though no record exists — not an error
  assert!(gc.is_black(&phantom_hash));
}

#[test]
fn record_with_empty_referenced_hashes() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();

  // WithRefs with empty children vec
  let leaf = insert_test(
    &stores,
    make_record_with_refs(vec![]),
    GenerationEpoch::new(0),
  );

  gc.start_marking(std::iter::once(test_ref(leaf)));
  mark_to_completion(&gc, &stores);

  assert!(gc.is_black(&leaf));
  assert_eq!(gc.gray_queue_len(), 0);
}

#[test]
fn deep_linear_chain() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  let r5 = insert_test(&stores, make_record("r5"), epoch);
  let r4 = insert_test(&stores, make_record_with_refs(vec![r5]), epoch);
  let r3 = insert_test(&stores, make_record_with_refs(vec![r4]), epoch);
  let r2 = insert_test(&stores, make_record_with_refs(vec![r3]), epoch);
  let r1 = insert_test(&stores, make_record_with_refs(vec![r2]), epoch);

  gc.start_marking(std::iter::once(test_ref(r1)));
  mark_to_completion(&gc, &stores);

  for h in &[r1, r2, r3, r4, r5] {
    assert!(gc.is_black(h), "hash should be black");
  }
}

#[test]
fn diamond_dependency() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  let r4 = insert_test(&stores, make_record("r4"), epoch);
  // Give r2 and r3 distinct extra leaves so they hash differently
  let r2_leaf = insert_test(&stores, make_record("r2_unique"), epoch);
  let r3_leaf = insert_test(&stores, make_record("r3_unique"), epoch);
  let r2 = insert_test(&stores, make_record_with_refs(vec![r4, r2_leaf]), epoch);
  let r3 = insert_test(&stores, make_record_with_refs(vec![r4, r3_leaf]), epoch);
  let r1 = insert_test(&stores, make_record_with_refs(vec![r2, r3]), epoch);

  gc.start_marking(std::iter::once(test_ref(r1)));
  mark_to_completion(&gc, &stores);

  for h in &[r1, r2, r3, r4, r2_leaf, r3_leaf] {
    assert!(gc.is_black(h));
  }
  // r4 should only appear once in black set despite two paths
  assert_eq!(gc.black_set_len(), 6);
}

#[test]
fn cycle_of_two() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  // We need to know hashes ahead of time. Create r2 first, then r1 refs r2,
  // then overwrite r2 to ref r1.
  let r2_initial = make_record("r2_initial");
  let r2_hash = hash_of(&r2_initial);
  insert_test(&stores, r2_initial, epoch);

  let r1 = make_record_with_refs(vec![r2_hash]);
  let r1_hash = hash_of(&r1);
  insert_test(&stores, r1, epoch);

  // Overwrite r2 in the store to point back at r1
  let r2_cyclic = make_record_with_refs(vec![r1_hash]);
  let _r2_cyclic_hash = hash_of(&r2_cyclic);
  insert_test(&stores, r2_cyclic, epoch);

  gc.start_marking(std::iter::once(test_ref(r1_hash)));
  mark_to_completion(&gc, &stores);

  assert!(gc.is_black(&r1_hash));
  assert!(gc.is_black(&r2_hash));
}

#[test]
fn self_referencing_record() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  let sentinel = ContentHash::new(&[42, 42, 42]);
  let record = make_record_with_refs(vec![sentinel]);
  let actual_hash = insert_test(&stores, record, epoch);

  // Root is actual_hash, it references sentinel which doesn't exist
  gc.start_marking(std::iter::once(test_ref(actual_hash)));
  mark_to_completion(&gc, &stores);

  assert!(gc.is_black(&actual_hash));
  assert!(gc.is_black(&sentinel));
}

#[test]
fn disconnected_subgraphs() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  // Rooted subgraph
  let b = insert_test(&stores, make_record("b"), epoch);
  let a = insert_test(&stores, make_record_with_refs(vec![b]), epoch);

  // Unreachable subgraph
  let d = insert_test(&stores, make_record("d"), epoch);
  let c = insert_test(&stores, make_record_with_refs(vec![d]), epoch);

  gc.start_marking(std::iter::once(test_ref(a)));
  mark_to_completion(&gc, &stores);

  assert!(gc.is_black(&a));
  assert!(gc.is_black(&b));
  assert!(!gc.is_black(&c));
  assert!(!gc.is_black(&d));
}

#[test]
fn wide_fan_out() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  let mut children = Vec::new();
  for i in 0..20u32 {
    let child = insert_test(
      &stores,
      make_record(&format!("child_{i}")),
      epoch,
    );
    children.push(child);
  }

  let root = insert_test(&stores, make_record_with_refs(children.clone()), epoch);

  gc.start_marking(std::iter::once(test_ref(root)));
  mark_to_completion(&gc, &stores);

  assert!(gc.is_black(&root));
  for child in &children {
    assert!(gc.is_black(child));
  }
  assert_eq!(gc.black_set_len(), 21);
}

#[test]
fn sweep_removes_unreferenced_old_record() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();

  // Old epoch record, not a root
  let orphan = insert_test(&stores, make_record("orphan"), GenerationEpoch::new(0));

  // Start GC — no roots
  gc.start_marking(std::iter::empty());
  mark_to_completion(&gc, &stores);
  gc.finish_marking();

  gc.sweep_partition::<TestPartitions, TestPartition>(&stores);
  gc.finish_sweep();

  assert!(!store_contains(&stores, &orphan));
}

#[test]
fn sweep_preserves_black_record() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();

  let live = insert_test(&stores, make_record("live"), GenerationEpoch::new(0));

  gc.start_marking(std::iter::once(test_ref(live)));
  mark_to_completion(&gc, &stores);
  gc.finish_marking();

  gc.sweep_partition::<TestPartitions, TestPartition>(&stores);
  gc.finish_sweep();

  assert!(store_contains(&stores, &live));
}

#[test]
fn sweep_preserves_new_epoch_record() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();

  // Start GC — epoch goes to 1
  gc.start_marking(std::iter::empty());
  let gc_epoch = gc.epoch();

  // Insert record at GC epoch (or later)
  let new_record = insert_test(&stores, make_record("new"), gc_epoch);

  mark_to_completion(&gc, &stores);
  gc.finish_marking();

  // new_record is NOT black, but has epoch >= gc_epoch, so survives
  assert!(!gc.is_black(&new_record));

  gc.sweep_partition::<TestPartitions, TestPartition>(&stores);
  gc.finish_sweep();

  assert!(store_contains(&stores, &new_record));
}

#[test]
fn sweep_mixed() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let old_epoch = GenerationEpoch::new(0);

  let live = insert_test(&stores, make_record("live"), old_epoch);
  let garbage = insert_test(&stores, make_record("garbage"), old_epoch);

  gc.start_marking(std::iter::once(test_ref(live)));
  let gc_epoch = gc.epoch();

  // Insert a new-epoch record during marking
  let new_epoch_record = insert_test(&stores, make_record("new_epoch"), gc_epoch);

  mark_to_completion(&gc, &stores);
  gc.finish_marking();

  gc.sweep_partition::<TestPartitions, TestPartition>(&stores);
  gc.finish_sweep();

  assert!(store_contains(&stores, &live));
  assert!(!store_contains(&stores, &garbage));
  assert!(store_contains(&stores, &new_epoch_record));
}

#[test]
fn sweep_all_garbage() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let old_epoch = GenerationEpoch::new(0);

  insert_test(&stores, make_record("g1"), old_epoch);
  insert_test(&stores, make_record("g2"), old_epoch);
  insert_test(&stores, make_record("g3"), old_epoch);

  assert_eq!(store_len(&stores), 3);

  gc.start_marking(std::iter::empty());
  mark_to_completion(&gc, &stores);
  gc.finish_marking();

  gc.sweep_partition::<TestPartitions, TestPartition>(&stores);
  gc.finish_sweep();

  assert_eq!(store_len(&stores), 0);
}

#[test]
fn sweep_empty_partition() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();

  assert_eq!(store_len(&stores), 0);

  gc.start_marking(std::iter::empty());
  mark_to_completion(&gc, &stores);
  gc.finish_marking();

  // Should not panic or error
  gc.sweep_partition::<TestPartitions, TestPartition>(&stores);
  gc.finish_sweep();

  assert_eq!(store_len(&stores), 0);
}

#[test]
fn add_to_gray_during_marking() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  let r1 = insert_test(&stores, make_record("r1"), epoch);

  gc.start_marking(std::iter::empty());

  // Write barrier: add r1 to gray
  gc.add_to_gray(std::iter::once(test_ref(r1)));
  assert_eq!(gc.gray_queue_len(), 1);

  mark_to_completion(&gc, &stores);

  assert!(gc.is_black(&r1));
}

#[test]
fn add_to_gray_already_black() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  let r1 = insert_test(&stores, make_record("r1"), epoch);

  gc.start_marking(std::iter::once(test_ref(r1)));
  mark_to_completion(&gc, &stores);

  // r1 is now black; adding to gray should be skipped
  let gray_before = gc.gray_queue_len();
  gc.add_to_gray(std::iter::once(test_ref(r1)));
  assert_eq!(gc.gray_queue_len(), gray_before);
}

#[test]
fn add_to_gray_when_idle() {
  let gc = GarbageCollector::new();
  let hash = ContentHash::new(&[1, 2, 3]);

  gc.add_to_gray(std::iter::once(test_ref(hash)));
  assert_eq!(gc.gray_queue_len(), 0);
}

#[test]
fn add_to_gray_when_sweeping() {
  let gc = GarbageCollector::new();
  gc.start_marking(std::iter::empty());
  gc.finish_marking();
  assert_eq!(gc.phase(), GcPhase::Sweeping);

  let hash = ContentHash::new(&[4, 5, 6]);
  gc.add_to_gray(std::iter::once(test_ref(hash)));
  assert_eq!(gc.gray_queue_len(), 0);
}

#[test]
fn write_barrier_saves_new_reference_chain() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  // Start marking with no roots
  gc.start_marking(std::iter::empty());
  let gc_epoch = gc.epoch();

  // During marking, add a reference chain
  let child = insert_test(&stores, make_record("child"), gc_epoch);
  let parent = insert_test(
    &stores,
    make_record_with_refs(vec![child]),
    gc_epoch,
  );

  // Write barrier adds parent to gray
  gc.add_to_gray(std::iter::once(test_ref(parent)));
  mark_to_completion(&gc, &stores);

  assert!(gc.is_black(&parent));
  assert!(gc.is_black(&child));

  gc.finish_marking();
  gc.sweep_partition::<TestPartitions, TestPartition>(&stores);
  gc.finish_sweep();

  // Both survive
  assert!(store_contains(&stores, &parent));
  assert!(store_contains(&stores, &child));
}

#[test]
fn budget_limits_processing() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  let mut hashes = Vec::new();
  for i in 0..10u32 {
    let h = insert_test(&stores, make_record(&format!("rec_{i}")), epoch);
    hashes.push(h);
  }

  gc.start_marking(hashes.iter().copied().map(test_ref));

  // Process with budget of 3
  gc.mark_tick_partition::<TestPartitions, TestPartition>(&stores, 3);

  // Should have processed at most 3
  assert!(gc.black_set_len() <= 3);
}

#[test]
fn budget_with_cascading_refs() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  // Chain: r1 -> r2 -> r3 -> r4 -> r5
  let r5 = insert_test(&stores, make_record("r5"), epoch);
  let r4 = insert_test(&stores, make_record_with_refs(vec![r5]), epoch);
  let r3 = insert_test(&stores, make_record_with_refs(vec![r4]), epoch);
  let r2 = insert_test(&stores, make_record_with_refs(vec![r3]), epoch);
  let r1 = insert_test(&stores, make_record_with_refs(vec![r2]), epoch);

  gc.start_marking(std::iter::once(test_ref(r1)));

  // Budget 2: process r1 and discover r2. Process r2 and discover r3.
  gc.mark_tick_partition::<TestPartitions, TestPartition>(&stores, 2);
  assert!(gc.black_set_len() >= 1 && gc.black_set_len() <= 2);

  // Continue with more ticks
  for _ in 0..10 {
    gc.mark_tick_partition::<TestPartitions, TestPartition>(&stores, 2);
  }

  // All should eventually be black
  for h in &[r1, r2, r3, r4, r5] {
    assert!(gc.is_black(h));
  }
}

#[test]
fn budget_zero() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  let r1 = insert_test(&stores, make_record("r1"), epoch);

  gc.start_marking(std::iter::once(test_ref(r1)));

  let done = gc.mark_tick_partition::<TestPartitions, TestPartition>(&stores, 0);

  // Nothing processed, gray not empty so not done
  assert!(!done);
  assert_eq!(gc.black_set_len(), 0);
}

#[test]
fn multiple_ticks_drain_queue() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  // Chain of 10
  let mut prev = insert_test(&stores, make_record("leaf"), epoch);
  for _i in 0..9u32 {
    prev = insert_test(
      &stores,
      make_record_with_refs(vec![prev]),
      epoch,
    );
  }

  gc.start_marking(std::iter::once(test_ref(prev)));

  // Drain with budget of 1 per tick
  let mut ticks = 0;
  loop {
    let done = gc.mark_tick_partition::<TestPartitions, TestPartition>(
      &stores,
      1,
    );
    ticks += 1;
    if done {
      break;
    }
    if ticks > 100 {
      panic!("failed to drain in 100 ticks");
    }
  }

  assert_eq!(gc.black_set_len(), 10);
}

#[test]
fn gc_mark_tick_processes_all_partitions() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  let h1 = insert_test(&stores, make_record("in_test"), epoch);
  let h2 = insert_test1(&stores, make_record("in_test1"), epoch);

  gc.start_marking(vec![test_ref(h1), test1_ref(h2)].into_iter());

  let done = TestPartitions::gc_mark_tick(&gc, &stores, 100);

  assert!(done);
  assert!(gc.is_black(&h1));
  assert!(gc.is_black(&h2));
}

#[test]
fn gc_sweep_cleans_all_partitions() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let old_epoch = GenerationEpoch::new(0);

  insert_test(&stores, make_record("garbage_test"), old_epoch);
  insert_test1(&stores, make_record("garbage_test1"), old_epoch);

  assert_eq!(store_len(&stores), 1);
  assert_eq!(store1_len(&stores), 1);

  gc.start_marking(std::iter::empty());
  let done = TestPartitions::gc_mark_tick(&gc, &stores, 100);
  assert!(done);
  gc.finish_marking();

  TestPartitions::gc_sweep(&gc, &stores);
  gc.finish_sweep();

  assert_eq!(store_len(&stores), 0);
  assert_eq!(store1_len(&stores), 0);
}

#[test]
fn cross_partition_reference() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  // Record in Test1Partition
  let child = insert_test1(&stores, make_record("child_in_test1"), epoch);

  // Parent in TestPartition references the child in Test1Partition
  let parent = insert_test(
    &stores,
    make_record_with_refs(vec![child]),
    epoch,
  );

  gc.start_marking(std::iter::once(test_ref(parent)));

  // Multiple ticks to ensure cross-partition traversal completes
  for _ in 0..10 {
    let done = TestPartitions::gc_mark_tick(&gc, &stores, 100);
    if done {
      break;
    }
  }

  assert!(gc.is_black(&parent));
  assert!(gc.is_black(&child));
}

#[test]
fn cross_partition_reference_reverse_direction() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  // Child in TestPartition
  let child = insert_test(&stores, make_record("child_in_test"), epoch);

  // Parent in Test1Partition references the child.
  // Test1Partition store is checked AFTER TestPartition in gc_mark_tick.
  let parent = insert_test1(
    &stores,
    make_record_with_refs(vec![child]),
    epoch,
  );

  gc.start_marking(std::iter::once(test1_ref(parent)));

  for _ in 0..10 {
    let done = TestPartitions::gc_mark_tick(&gc, &stores, 100);
    if done {
      break;
    }
  }

  assert!(gc.is_black(&parent));
  assert!(gc.is_black(&child));
}

#[test]
fn full_cycle_collects_garbage() {
  let db: Database<TestPartitions> = Database::new();
  let source_cache =
    crate::source::cache::reporter::SourceCacheReader::new_empty_for_test();
  let gc = GarbageCollector::new();

  // Commit record1 under key "a"
  let record1 = make_record("first");
  let hash1 = hash_of(&record1);
  let mut writer1 = RecordWriter::<TestPartitions>::new(Ident::new("t1"));
  writer1.insert::<TestPartition, _>("a".to_string(), record1);
  let _ = db.commit_chunk(writer1.build(), &source_cache);

  // Overwrite key "a" with record2 — record1 becomes garbage in CAS
  let record2 = make_record("second");
  let hash2 = hash_of(&record2);
  let mut writer2 = RecordWriter::<TestPartitions>::new(Ident::new("t2"));
  writer2.insert::<TestPartition, _>("a".to_string(), record2);
  let _ = db.commit_chunk(writer2.build(), &source_cache);

  let stores = db.get_store().stores();

  let roots = db.collect_index_hashes();
  assert!(roots.iter().any(|r| r.hash == hash2));

  gc.start_marking(roots.into_iter());
  mark_to_completion(&gc, stores);
  gc.finish_marking();
  TestPartitions::gc_sweep(&gc, stores);
  gc.finish_sweep();

  assert!(store_contains(stores, &hash2));
  assert!(!store_contains(stores, &hash1));
}

#[test]
fn two_consecutive_cycles() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();

  let live = insert_test(&stores, make_record("live"), GenerationEpoch::new(0));

  // Cycle 1: insert orphan at the GC epoch so it's protected
  gc.start_marking(std::iter::once(test_ref(live)));
  let gc_epoch_1 = gc.epoch();
  let orphan = insert_test(&stores, make_record("orphan"), gc_epoch_1);
  mark_to_completion(&gc, &stores);
  gc.finish_marking();
  gc.sweep_partition::<TestPartitions, TestPartition>(&stores);
  gc.finish_sweep();

  // orphan survived (epoch >= gc_epoch)
  assert!(store_contains(&stores, &orphan));

  // Cycle 2: orphan's epoch is now old relative to new gc_epoch
  gc.start_marking(std::iter::once(test_ref(live)));
  mark_to_completion(&gc, &stores);
  gc.finish_marking();
  gc.sweep_partition::<TestPartitions, TestPartition>(&stores);
  gc.finish_sweep();

  // orphan now collected (not black, epoch < gc_epoch_2)
  assert!(!store_contains(&stores, &orphan));
  assert!(store_contains(&stores, &live));
}

#[test]
fn cycle_with_no_garbage() {
  let stores = TestPartitions::new_stores();
  let gc = GarbageCollector::new();
  let epoch = GenerationEpoch::new(0);

  let r1 = insert_test(&stores, make_record("r1"), epoch);
  let r2 = insert_test(&stores, make_record("r2"), epoch);
  let r3 = insert_test(&stores, make_record("r3"), epoch);

  gc.start_marking(vec![test_ref(r1), test_ref(r2), test_ref(r3)].into_iter());
  mark_to_completion(&gc, &stores);
  gc.finish_marking();

  gc.sweep_partition::<TestPartitions, TestPartition>(&stores);
  gc.finish_sweep();

  assert_eq!(store_len(&stores), 3);
  assert!(store_contains(&stores, &r1));
  assert!(store_contains(&stores, &r2));
  assert!(store_contains(&stores, &r3));
}