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

Add local KV store commands

Browse source
This commit is contained in:
Eric Wendland 2026-05-16 21:52:35 +02:00
commit 1a0f1e7671
14 changed files with 395 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -407,9 +407,10 @@ fn request_for_command(command: Command) -> Result<ControlRequest> {
CasCommand::Cleanup { dry_run } => ControlRequest::CasCleanup { dry_run },
CasCommand::List => ControlRequest::CasList,
},
Command::Kv { command } => ControlRequest::ModuleStub {
module: "kv".to_owned(),
command: format!("{command:?}"),
Command::Kv { command } => match command {
KvCommand::Create { name } => ControlRequest::KvCreate { name },
KvCommand::Set { name, key, value } => ControlRequest::KvSet { name, key, value },
KvCommand::Get { name, key } => ControlRequest::KvGet { name, key },
},
Command::Pubsub { command } => ControlRequest::ModuleStub {
module: "pubsub".to_owned(),
@ -766,6 +767,22 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> {
println!("schema_metadata: {}", db.schema_metadata);
println!("sync_status: {}", db.sync_status);
}
ControlResponse::KvCreated { kv } => {
println!("created kv: {}", kv.name);
println!("id: {}", kv.id);
println!("resource: {}", kv.resource);
println!("sync_status: {}", kv.sync_status);
}
ControlResponse::KvSet { entry } => {
println!("set {} {}", entry.store, entry.key);
}
ControlResponse::KvGet { entry } => {
if let Some(entry) = entry {
println!("{}", entry.value);
} else {
println!("not found");
}
}
ControlResponse::NotImplemented { module, command } => {
println!("{module} {command}: not implemented yet");
}

1
crates/geth-control/Cargo.toml
View file

@ -12,6 +12,7 @@ thiserror.workspace = true
geth-auth = { path = "../geth-auth" }
geth-db = { path = "../geth-db" }
geth-keychain = { path = "../geth-keychain" }
geth-kv = { path = "../geth-kv" }
geth-resource = { path = "../geth-resource" }
geth-ssh-identity = { path = "../geth-ssh-identity" }
geth-types = { path = "../geth-types" }

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

@ -1,6 +1,7 @@
use geth_auth::{AuthExplanation, AuthOp};
use geth_db::DbResource;
use geth_keychain::KeychainOp;
use geth_kv::{KvEntry, KvResource};
use geth_resource::ResourceDescriptor;
use geth_ssh_identity::{
SshCertApproval, SshCertRequest, SshCertificateRecord, SshRevocationEntry,
@ -98,6 +99,18 @@ pub enum ControlRequest {
DbStatus {
name: String,
},
KvCreate {
name: String,
},
KvSet {
name: String,
key: String,
value: String,
},
KvGet {
name: String,
key: String,
},
ModuleStub {
module: String,
command: String,
@ -183,6 +196,15 @@ pub enum ControlResponse {
DbStatus {
db: DbResource,
},
KvCreated {
kv: KvResource,
},
KvSet {
entry: KvEntry,
},
KvGet {
entry: Option<KvEntry>,
},
NotImplemented {
module: String,
command: String,

1
crates/geth-kv/Cargo.toml
View file

@ -7,4 +7,5 @@ license.workspace = true
[dependencies]
serde.workspace = true
thiserror.workspace = true
geth-types = { path = "../geth-types" }

57
crates/geth-kv/src/lib.rs
View file

@ -6,9 +6,66 @@ pub struct KvResource {
pub id: KvId,
pub resource: ResourceId,
pub name: String,
pub sync_status: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct KvEntry {
pub store: KvId,
pub key: String,
pub value: String,
}
#[derive(Debug, thiserror::Error)]
pub enum KvError {
#[error("invalid kv store name: {0}")]
InvalidName(String),
#[error("invalid kv key: {0}")]
InvalidKey(String),
}
pub fn validate_kv_name(name: &str) -> Result<(), KvError> {
if name.is_empty()
|| !name
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
{
return Err(KvError::InvalidName(name.to_owned()));
}
Ok(())
}
pub fn validate_kv_key(key: &str) -> Result<(), KvError> {
if key.is_empty() || key.bytes().any(|byte| byte == 0 || byte == b'\n') {
return Err(KvError::InvalidKey(key.to_owned()));
}
Ok(())
}
#[must_use]
pub fn iroh_docs_roadmap() -> &'static str {
"future kv storage uses Iroh Documents namespaces with prefix-scoped authorization"
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn kv_name_validation_rejects_paths_and_empty_names() {
assert!(validate_kv_name("prefs").is_ok());
assert!(validate_kv_name("apps.foo").is_ok());
assert!(validate_kv_name("").is_err());
assert!(validate_kv_name("../prefs").is_err());
assert!(validate_kv_name("prefs/main").is_err());
assert!(validate_kv_name("prefs main").is_err());
}
#[test]
fn kv_key_validation_rejects_empty_newline_and_nul() {
assert!(validate_kv_key("apps/foo/theme").is_ok());
assert!(validate_kv_key("").is_err());
assert!(validate_kv_key("apps/foo\nbar").is_err());
assert!(validate_kv_key("apps/foo\0bar").is_err());
}
}

1
crates/geth-node/Cargo.toml
View file

@ -18,6 +18,7 @@ geth-crypto = { path = "../geth-crypto" }
geth-db = { path = "../geth-db" }
geth-iroh = { path = "../geth-iroh" }
geth-keychain = { path = "../geth-keychain" }
geth-kv = { path = "../geth-kv" }
geth-resource = { path = "../geth-resource" }
geth-ssh-identity = { path = "../geth-ssh-identity" }
geth-store = { path = "../geth-store" }

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

@ -11,6 +11,7 @@ use geth_crypto::AgentKey;
use geth_db::DbResource;
use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode};
use geth_keychain::{KeychainOp, KeychainOpKind};
use geth_kv::{KvEntry, KvResource};
use geth_resource::ResourceDescriptor;
use geth_ssh_identity::{
SshCertApproval, SshCertKind, SshCertRequest, SshCertRequestStatus, SshCertificateRecord,
@ -18,8 +19,8 @@ use geth_ssh_identity::{
certificate_id, revocation_id, ssh_public_key_fingerprint,
};
use geth_store::{
Store, StoredAuthOp, StoredDbResource, StoredKeychainOp, StoredResource, StoredSshCertRequest,
StoredSshCertificate, StoredSshRevocation,
Store, StoredAuthOp, StoredDbResource, StoredKeychainOp, StoredKvEntry, StoredKvStore,
StoredResource, StoredSshCertRequest, StoredSshCertificate, StoredSshRevocation,
};
use geth_types::{
AuthOpId, Capability, KeyId, NodeId, PrincipalId, ResourceId, ResourceKind, ResourceName,
@ -53,6 +54,12 @@ pub enum NodeError {
InvalidDbPath(String),
#[error("db resource not found: {0}")]
DbNotFound(String),
#[error("invalid kv store name: {0}")]
InvalidKvName(String),
#[error("invalid kv key: {0}")]
InvalidKvKey(String),
#[error("kv store not found: {0}")]
KvNotFound(String),
#[error("invalid ssh certificate kind: {0}")]
InvalidSshCertKind(String),
#[error("invalid ssh certificate request status: {0}")]
@ -592,6 +599,55 @@ pub fn handle_request(
db: db_resource_from_stored(&stored)?,
})
}
ControlRequest::KvCreate { name } => {
geth_kv::validate_kv_name(&name).map_err(|_| NodeError::InvalidKvName(name.clone()))?;
let resource_id = format!("resource:kv:{name}");
let kv_id = format!("kv:{name}");
let resource = StoredResource {
resource_id: resource_id.clone(),
kind: ResourceKind::Kv.to_string(),
name: name.clone(),
status: "active".to_owned(),
};
store.insert_resource(&resource)?;
let stored = StoredKvStore {
kv_id,
resource_id,
name,
};
store.insert_kv_store(&stored)?;
Ok(ControlResponse::KvCreated {
kv: kv_resource_from_stored(&stored),
})
}
ControlRequest::KvSet { name, key, value } => {
geth_kv::validate_kv_name(&name).map_err(|_| NodeError::InvalidKvName(name.clone()))?;
geth_kv::validate_kv_key(&key).map_err(|_| NodeError::InvalidKvKey(key.clone()))?;
let kv = store
.get_kv_store_by_name(&name)?
.ok_or_else(|| NodeError::KvNotFound(name.clone()))?;
let stored = StoredKvEntry {
kv_id: kv.kv_id,
key,
value,
updated_at_ms: geth_store::now_ms(),
};
store.set_kv_entry(&stored)?;
Ok(ControlResponse::KvSet {
entry: kv_entry_from_stored(stored),
})
}
ControlRequest::KvGet { name, key } => {
geth_kv::validate_kv_name(&name).map_err(|_| NodeError::InvalidKvName(name.clone()))?;
geth_kv::validate_kv_key(&key).map_err(|_| NodeError::InvalidKvKey(key.clone()))?;
let kv = store
.get_kv_store_by_name(&name)?
.ok_or_else(|| NodeError::KvNotFound(name.clone()))?;
let entry = store
.get_kv_entry(&kv.kv_id, &key)?
.map(kv_entry_from_stored);
Ok(ControlResponse::KvGet { entry })
}
ControlRequest::ModuleStub { module, command } => {
Ok(ControlResponse::NotImplemented { module, command })
}
@ -625,6 +681,23 @@ fn db_resource_from_stored(stored: &StoredDbResource) -> Result<DbResource, Node
})
}
fn kv_resource_from_stored(stored: &StoredKvStore) -> KvResource {
KvResource {
id: stored.kv_id.clone().into(),
resource: stored.resource_id.clone().into(),
name: stored.name.clone(),
sync_status: "local-only".to_owned(),
}
}
fn kv_entry_from_stored(stored: StoredKvEntry) -> KvEntry {
KvEntry {
store: stored.kv_id.into(),
key: stored.key,
value: stored.value,
}
}
fn store_auth_op(store: &Store, op: &AuthOp) -> Result<(), NodeError> {
store.insert_auth_op(&StoredAuthOp {
op_id: op.id.to_string(),

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

@ -105,6 +105,13 @@ impl Store {
resource_id TEXT NOT NULL,
name TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS kv_entries (
kv_id TEXT NOT NULL,
key TEXT NOT NULL,
value TEXT NOT NULL,
updated_at_ms INTEGER NOT NULL,
PRIMARY KEY(kv_id, key)
);
CREATE TABLE IF NOT EXISTS db_resources (
db_id TEXT PRIMARY KEY,
resource_id TEXT NOT NULL,
@ -275,6 +282,63 @@ impl Store {
}
}
pub fn insert_kv_store(&self, kv: &StoredKvStore) -> Result<(), StoreError> {
self.conn.execute(
r#"INSERT OR REPLACE INTO kv_stores(kv_id, resource_id, name)
VALUES (?1, ?2, ?3)"#,
params![kv.kv_id, kv.resource_id, kv.name],
)?;
Ok(())
}
pub fn get_kv_store_by_name(&self, name: &str) -> Result<Option<StoredKvStore>, StoreError> {
let mut stmt = self.conn.prepare(
r#"SELECT kv_id, resource_id, name
FROM kv_stores WHERE name = ?1"#,
)?;
let mut rows = stmt.query(params![name])?;
if let Some(row) = rows.next()? {
Ok(Some(StoredKvStore {
kv_id: row.get(0)?,
resource_id: row.get(1)?,
name: row.get(2)?,
}))
} else {
Ok(None)
}
}
pub fn set_kv_entry(&self, entry: &StoredKvEntry) -> Result<(), StoreError> {
self.conn.execute(
r#"INSERT OR REPLACE INTO kv_entries(kv_id, key, value, updated_at_ms)
VALUES (?1, ?2, ?3, ?4)"#,
params![entry.kv_id, entry.key, entry.value, entry.updated_at_ms],
)?;
Ok(())
}
pub fn get_kv_entry(
&self,
kv_id: &str,
key: &str,
) -> Result<Option<StoredKvEntry>, StoreError> {
let mut stmt = self.conn.prepare(
r#"SELECT kv_id, key, value, updated_at_ms
FROM kv_entries WHERE kv_id = ?1 AND key = ?2"#,
)?;
let mut rows = stmt.query(params![kv_id, key])?;
if let Some(row) = rows.next()? {
Ok(Some(StoredKvEntry {
kv_id: row.get(0)?,
key: row.get(1)?,
value: row.get(2)?,
updated_at_ms: row.get(3)?,
}))
} else {
Ok(None)
}
}
pub fn record_cas_object(
&self,
hash: &str,
@ -622,6 +686,21 @@ pub struct StoredDbResource {
pub path: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredKvStore {
pub kv_id: String,
pub resource_id: String,
pub name: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredKvEntry {
pub kv_id: String,
pub key: String,
pub value: String,
pub updated_at_ms: i64,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CasObject {
pub hash: String,
@ -883,4 +962,33 @@ mod tests {
Some("/tmp/notes.sqlite".to_owned())
);
}
#[test]
fn kv_store_and_entries_roundtrip() {
let store = Store::open_memory().expect("open");
let kv = StoredKvStore {
kv_id: "kv:prefs".to_owned(),
resource_id: "resource:kv:prefs".to_owned(),
name: "prefs".to_owned(),
};
store.insert_kv_store(&kv).expect("insert kv");
assert_eq!(
store.get_kv_store_by_name("prefs").expect("get kv"),
Some(kv.clone())
);
let entry = StoredKvEntry {
kv_id: kv.kv_id.clone(),
key: "apps/foo/theme".to_owned(),
value: "dark".to_owned(),
updated_at_ms: 1,
};
store.set_kv_entry(&entry).expect("set entry");
assert_eq!(
store
.get_kv_entry("kv:prefs", "apps/foo/theme")
.expect("get entry"),
Some(entry)
);
}
}

91
crates/geth/tests/bootstrap.rs
View file

@ -454,6 +454,97 @@ fn db_add_and_status_register_local_db_metadata() {
);
}
#[test]
fn kv_create_set_get_use_local_store() {
let home = tempfile::tempdir().expect("tempdir");
let paths = geth_config::GethPaths::from_home(home.path());
let node = geth_node::init_node(&paths).expect("init node");
let response = geth_node::handle_request(
&node,
geth_control::ControlRequest::KvCreate {
name: "prefs".to_owned(),
},
)
.expect("create kv");
match response {
geth_control::ControlResponse::KvCreated { kv } => {
assert_eq!(kv.name, "prefs");
assert_eq!(kv.sync_status, "local-only");
}
other => panic!("unexpected response: {other:?}"),
}
let response = geth_node::handle_request(
&node,
geth_control::ControlRequest::KvSet {
name: "prefs".to_owned(),
key: "apps/foo/theme".to_owned(),
value: "dark".to_owned(),
},
)
.expect("set kv");
match response {
geth_control::ControlResponse::KvSet { entry } => {
assert_eq!(entry.store.to_string(), "kv:prefs");
assert_eq!(entry.key, "apps/foo/theme");
assert_eq!(entry.value, "dark");
}
other => panic!("unexpected response: {other:?}"),
}
let response = geth_node::handle_request(
&node,
geth_control::ControlRequest::KvGet {
name: "prefs".to_owned(),
key: "apps/foo/theme".to_owned(),
},
)
.expect("get kv");
match response {
geth_control::ControlResponse::KvGet { entry } => {
assert_eq!(entry.expect("entry").value, "dark");
}
other => panic!("unexpected response: {other:?}"),
}
let response = geth_node::handle_request(
&node,
geth_control::ControlRequest::KvGet {
name: "prefs".to_owned(),
key: "apps/foo/missing".to_owned(),
},
)
.expect("get missing kv");
match response {
geth_control::ControlResponse::KvGet { entry } => {
assert!(entry.is_none());
}
other => panic!("unexpected response: {other:?}"),
}
assert!(
geth_node::handle_request(
&node,
geth_control::ControlRequest::KvCreate {
name: "../bad".to_owned(),
},
)
.is_err()
);
assert!(
geth_node::handle_request(
&node,
geth_control::ControlRequest::KvSet {
name: "missing".to_owned(),
key: "apps/foo/theme".to_owned(),
value: "dark".to_owned(),
},
)
.is_err()
);
}
#[test]
fn ssh_cert_request_approval_and_revocation_export_use_local_state() {
let home = tempfile::tempdir().expect("tempdir");