Function collect 

pub async fn collect(
    store: &dyn Store,
    prefix: &KeyPrefix,
    horizon: Duration,
    now_ms: i64,
) -> Result<CollectReport>

Purge tombstones under prefix whose deleted_at is at least horizon old.

Live records are never touched. A tombstone with no deleted_at is kept and counted in CollectReport::unstamped. A future-dated deleted_at (clock skew) gives a negative age and is kept, so skew can delay collection but can’t trigger it early. A horizon longer than any representable age purges nothing. now_ms is a parameter so tests control time; the CLI passes the system clock.

Stops at the first store error and returns it; tombstones already purged stay purged, so a re-run is safe. Over a daemon, purge needs an admin token: a non-admin token fails on the first eligible tombstone, before anything is purged. A daemon predating replication fails on list_raw with DAEMON_PREDATES_REPLICATION.