You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: cluster_init.rs
    path of file: ./repos/surrealdb/core/src/kvs/tests
    
    the code of the file until where you have to start completion: use futures::lock::Mutex;
use std::collections::BTreeSet;
use std::sync::Arc;

use crate::ctx::context;

use crate::dbs::{Options, Session};
use crate::iam::{Auth, Role};
use crate::kvs::lq_structs::{LqValue, UnreachableLqType};
use crate::kvs::{LockType::*, TransactionType::*};
use crate::sql;
use crate::sql::statements::LiveStatement;
use crate::sql::Value::Table;
use crate::sql::{Fields, Value};
use test_log::test;
use uuid;

#[tokio::test]
#[serial]
async fn expired_nodes_are_garbage_collected() {
	let old_node = Uuid::parse_str("2ea6d33f-4c0a-417a-ab04-1fa9869f9a65").unwrap();
	let new_node = Uuid::parse_str("fbfb3487-71fe-4749-b3aa-1cc0a5380cdd").unwrap();
	let old_time = Timestamp {
		value: 123000,
	};
	let fake_clock = FakeClock::new(old_time);
	let fake_clock = Arc::new(SizedClock::Fake(fake_clock));
	let mut test = init(new_node, fake_clock.clone()).await.unwrap();

	// Set up the first node at an early timestamp
	test.db = test.db.with_node_id(sql::Uuid::from(old_node));
	test.db.bootstrap().await.unwrap();

	// Throw in some stray nodes and heartbeats
	let mut tx = test.db.transaction(Write, Optimistic).await.unwrap();
	let corrupt_node_1 = Uuid::parse_str("5a65fe57-7ac3-4b13-a31f-6376d3b484c8").unwrap();
	let corrupt_node_2 = Uuid::parse_str("eb94a0b4-70ea-482f-a7dd-dc02132be846").unwrap();
	tx.set_nd(corrupt_node_1).await.unwrap();
	tx.set_hb(old_time, corrupt_node_2).await.unwrap();
	tx.commit().await.unwrap();

	// Set up second node at a later timestamp
	let new_time = Timestamp {
		value: 567000,
	};
	set_fake_clock(fake_clock.clone(), new_time).await;
	test.db = test.db.with_node_id(sql::Uuid::from(new_node));
	test.db.bootstrap().await.unwrap();

	// Now scan the heartbeats to validate there is only one node left
	let mut tx = test.db.transaction(Write, Optimistic).await.unwrap();
	let scanned = tx.scan_hb(&new_time, 100).await.unwrap();
	assert_eq!(scanned.len(), 1);
	for hb in scanned.iter() {
		assert_eq!(&hb.nd, &new_node);
	}

	// And scan the nodes to verify its just the latest also
	let scanned = tx.scan_nd(100).await.unwrap();
	assert_eq!(scanned.len(), 1);
	for cl in scanned.iter() {
		assert_eq!(&cl.name, &ne
    
async fn single_live_queries_are_garbage_collected() {
	// Test parameters
	let ctx = context::Context::background();
	let node_id = Uuid::parse_str("b1a08614-a826-4581-938d-bea17f00e253").unwrap();
	let time = Timestamp {
		value: 123000,
	};
	let fake_clock = FakeClock::new(time);
	let fake_clock = Arc::new(SizedClock::Fake(fake_clock));
	let mut test = init(node_id, fake_clock).await.unwrap();
	let namespace = "test_namespace";
	let database = "test_db";
	let table = "test_table";
	let options = Options::default()
		.with_required(
			node_id,
			Some(Arc::from(namespace)),
			Some(Arc::from(database)),
			Arc::new(Auth::for_root(Role::Owner)),
		)
		.with_live(true);

	// We do standard cluster init
	trace!("Bootstrapping node {}", node_id);
	test.db = test.db.with_node_id(crate::sql::uuid::Uuid::from(node_id));
	test.db.bootstrap().await.unwrap();

	// We set up 2 live queries, one of which we want to garbage collect
	trace!("Setting up live queries");
	let tx = Arc::new(Mutex::new(test.db.transaction(Write, Optimistic).await.unwrap()));
	let live_query_to_delete = Uuid::parse_str("8aed07c4-9683-480e-b1e4-f0db8b331530").unwrap();
	let live_st = LiveStatement {
		id: sql::Uuid(live_query_to_delete),
		node: sql::uuid::Uuid::from(node_id),
		expr: Fields(vec![sql::Field::All], false),
		what: Table(sql::Table::from(table)),
		cond: None,
		fetch: None,
		archived: None,
		session: Some(Value::None),
		auth: Some(Auth::for_root(Role::Owner)),
	};
	live_st
		.compute(&ctx, &options, &tx, None)
		.await
		.map_err(|e| format!("Error computing live statement: {:?} {:?}", live_st, e))
		.unwrap();
	let live_query_to_keep = Uuid::parse_str("adea762a-17db-4810-a4a2-c54babfdaf23").unwrap();
	let live_st = LiveStatement {
		id: sql::Uuid(live_query_to_keep),
		node: sql::Uuid::from(node_id),
		expr: Fields(vec![sql::Field::All], false),
		what: Table(sql::Table::from(table)),
		cond: None,
		fetch: None,
		archived: None,
		session: Some(Value::None),
		auth: Some(Auth::for_root(Role::Owner)),
	};
	live_st
		.compute(&ctx, &options, &tx, None)
		.await
		.map_err(|e| format!("Error computing live statement: {:?} {:?}", live_st, e))
		.unwrap();
	tx.lock().await.commit().await.unwrap();

	// Subject: Perform the action we are testing
	trace!("Garbage collecting dead sessions");
	test.db.garbage_collect_dead_session(&[live_query_to_delete]).await.unwrap();

	// Validate
	trace!("Validating live queries");
	let mut tx = test.db.transaction(Write, Optimistic).await.unwrap();
	let scanned = tx.all_tb_lives(namespace, database, table).await.unwrap();
	assert_eq!(scanned.len(), 1, "The scanned values are {:?}", scanned);
	assert_eq!(&scanned[0].id.0, &live_query_to_keep);
	let scanned = tx.all_lq(&node_id).await.unwrap();
	assert_eq!(scanned.len(), 1);
	assert_eq!(&scanned[0].lq, &sql::Uuid::from(live_query_to_keep));
	tx.commit().await.unwrap();
}