eric/geth
Watch
1
0
Fork
You've already forked geth
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Track CAS providers

Browse source
This commit is contained in:
Eric Wendland 2026-05-19 15:37:02 +02:00
commit c9803ea7f2
8 changed files with 161 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

14
crates/geth-cli/src/lib.rs
View file

@ -243,6 +243,9 @@ pub enum CasCommand {
#[arg(long)]
dry_run: bool,
},
Providers {
hash: String,
},
List,
Root {
#[command(subcommand)]
@ -593,6 +596,7 @@ fn request_for_command(command: Command) -> Result<ControlRequest> {
CasCommand::Pin { hash } => ControlRequest::CasPin { hash: hash.into() },
CasCommand::Unpin { hash } => ControlRequest::CasUnpin { hash: hash.into() },
CasCommand::Cleanup { dry_run } => ControlRequest::CasCleanup { dry_run },
CasCommand::Providers { hash } => ControlRequest::CasProviders { hash: hash.into() },
CasCommand::List => ControlRequest::CasList,
CasCommand::Root { command } => match command {
CasRootCommand::Add { name, path } => ControlRequest::CasRootAdd { name, path },
@ -979,6 +983,16 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> {
println!("{}\t{} bytes\t{}", blob.hash, blob.size_bytes, pin);
}
}
ControlResponse::CasProviders { hash, providers } => {
println!("hash: {hash}");
println!("providers: {}", providers.len());
for provider in providers {
println!(
"{}\t{}\t{}",
provider.peer_node_id, provider.endpoint_id, provider.last_seen_ms
);
}
}
ControlResponse::CasRootAdded { root } => {
println!("added file root: {}", root.name);
println!("id: {}", root.id);

35
crates/geth-control/src/lib.rs
View file

@ -67,6 +67,9 @@ pub enum ControlRequest {
CasCleanup {
dry_run: bool,
},
CasProviders {
hash: BlobHash,
},
CasList,
CasRootAdd {
name: String,
@ -324,6 +327,10 @@ pub enum ControlResponse {
CasList {
blobs: Vec<CasBlob>,
},
CasProviders {
hash: BlobHash,
providers: Vec<CasProvider>,
},
CasRootAdded {
root: FileRoot,
},
@ -569,6 +576,13 @@ pub struct CasBlob {
pub pinned: bool,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CasProvider {
pub peer_node_id: String,
pub endpoint_id: String,
pub last_seen_ms: i64,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "kebab-case")]
pub enum PeerControlRequest {
@ -881,6 +895,27 @@ mod tests {
request
);
let request = ControlRequest::CasProviders {
hash: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".into(),
};
assert_eq!(
decode_request(&encode_request(&request).expect("encode")).expect("decode"),
request
);
let response = ControlResponse::CasProviders {
hash: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".into(),
providers: vec![CasProvider {
peer_node_id: "node:peer".to_owned(),
endpoint_id: "endpoint:peer".to_owned(),
last_seen_ms: 42,
}],
};
assert_eq!(
decode_response(&encode_response(&response).expect("encode")).expect("decode"),
response
);
let request = ControlRequest::SshRevocationExport {
out: PathBuf::from("revocations.krl-spec"),
format: "openssh-krl-spec".to_owned(),

24
crates/geth-node/src/lib.rs
View file

@ -8,7 +8,7 @@ use geth_cas::{
};
use geth_config::{GethConfig, GethPaths, RelayMode};
use geth_control::{
CasBlob, ControlRequest, ControlResponse, KeychainStatusResponse, NodeIdResponse,
CasBlob, CasProvider, ControlRequest, ControlResponse, KeychainStatusResponse, NodeIdResponse,
PeerControlRequest, PeerControlResponse, StatusResponse, SyncWatermark,
};
use geth_crypto::AgentKey;
@ -799,6 +799,7 @@ async fn cas_fetch_from_peer(
info.size_bytes,
&info.path.to_string_lossy(),
)?;
store.record_cas_provider(info.hash.as_str(), &node_id, &endpoint_id)?;
Ok(ControlResponse::CasFetched {
peer_node_id: node_id,
peer_agent_id: agent_id,
@ -2592,6 +2593,21 @@ pub fn handle_request(
dry_run,
})
}
ControlRequest::CasProviders { hash } => {
geth_cas::validate_hash(&hash)?;
Ok(ControlResponse::CasProviders {
providers: store
.list_cas_providers(hash.as_str())?
.into_iter()
.map(|provider| CasProvider {
peer_node_id: provider.peer_node_id,
endpoint_id: provider.endpoint_id,
last_seen_ms: provider.last_seen_ms,
})
.collect(),
hash,
})
}
ControlRequest::CasList => {
let cas = LocalCas::new(node.paths.cas_dir());
let blobs = cas
@ -4651,6 +4667,12 @@ mod tests {
}
other => panic!("unexpected allowed CAS fetch response: {other:?}"),
}
let providers = Store::open(&left_paths.metadata_db())
.expect("open left after cas fetch")
.list_cas_providers(right_blob.hash.as_str())
.expect("list cas providers");
assert_eq!(providers.len(), 1);
assert_eq!(providers[0].peer_node_id, right.node_id);
let kv_synced = handle_request_async(
&left,

72
crates/geth-store/src/lib.rs
View file

@ -100,6 +100,13 @@ impl Store {
hash TEXT PRIMARY KEY,
pinned_at_ms INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS cas_providers (
hash TEXT NOT NULL,
peer_node_id TEXT NOT NULL,
endpoint_id TEXT NOT NULL,
last_seen_ms INTEGER NOT NULL,
PRIMARY KEY(hash, peer_node_id)
);
CREATE TABLE IF NOT EXISTS kv_stores (
kv_id TEXT PRIMARY KEY,
resource_id TEXT NOT NULL,
@ -746,6 +753,39 @@ impl Store {
.map_err(StoreError::from)
}
pub fn record_cas_provider(
&self,
hash: &str,
peer_node_id: &str,
endpoint_id: &str,
) -> Result<(), StoreError> {
self.conn.execute(
r#"INSERT OR REPLACE INTO cas_providers(hash, peer_node_id, endpoint_id, last_seen_ms)
VALUES (?1, ?2, ?3, ?4)"#,
params![hash, peer_node_id, endpoint_id, now_ms()],
)?;
Ok(())
}
pub fn list_cas_providers(&self, hash: &str) -> Result<Vec<CasProvider>, StoreError> {
let mut stmt = self.conn.prepare(
r#"SELECT hash, peer_node_id, endpoint_id, last_seen_ms
FROM cas_providers
WHERE hash = ?1
ORDER BY last_seen_ms DESC, peer_node_id"#,
)?;
let rows = stmt.query_map(params![hash], |row| {
Ok(CasProvider {
hash: row.get(0)?,
peer_node_id: row.get(1)?,
endpoint_id: row.get(2)?,
last_seen_ms: row.get(3)?,
})
})?;
rows.collect::<Result<Vec<_>, _>>()
.map_err(StoreError::from)
}
pub fn upsert_peer_card(&self, peer_card: &StoredPeerCard) -> Result<(), StoreError> {
self.conn.execute(
"INSERT OR REPLACE INTO peer_cards(peer_id, card_json, updated_at_ms) VALUES (?1, ?2, ?3)",
@ -1132,6 +1172,14 @@ pub struct StoredResource {
pub status: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CasProvider {
pub hash: String,
pub peer_node_id: String,
pub endpoint_id: String,
pub last_seen_ms: i64,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredDbResource {
pub db_id: String,
@ -1452,6 +1500,30 @@ mod tests {
assert!(store.list_cas_pins().expect("pins").is_empty());
}
#[test]
fn cas_providers_roundtrip_by_hash() {
let store = Store::open_memory().expect("open");
let hash = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
store
.record_cas_provider(hash, "node:laptop", "endpoint:laptop")
.expect("record provider");
let providers = store.list_cas_providers(hash).expect("list providers");
assert_eq!(providers.len(), 1);
assert_eq!(providers[0].hash, hash);
assert_eq!(providers[0].peer_node_id, "node:laptop");
assert_eq!(providers[0].endpoint_id, "endpoint:laptop");
assert!(
store
.list_cas_providers(
"ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
)
.expect("list missing")
.is_empty()
);
}
#[test]
fn deleting_cas_object_removes_pin_metadata() {
let store = Store::open_memory().expect("open");