Add crsql changes metadata status

This commit is contained in:
Eric Wendland 2026-05-17 20:01:36 +02:00
commit 677945073b
8 changed files with 182 additions and 15 deletions

View file

@ -119,8 +119,9 @@ Roadmap items should be actionable and checkable:
- Local CAS supports pin/unpin metadata, surfaced through `cas list`, and - Local CAS supports pin/unpin metadata, surfaced through `cas list`, and
`cas cleanup` evicts unpinned blobs while retaining pinned blobs. `cas cleanup` evicts unpinned blobs while retaining pinned blobs.
- DB resources can be registered locally and report local-only status plus a - DB resources can be registered locally and report local-only status plus a
read-only SQLite schema summary/hash. cr-sqlite loading, change extraction, read-only SQLite schema summary/hash and `crsql_changes` metadata when
and sync are still roadmap work. present. cr-sqlite loading, typed change extraction, and sync are still
roadmap work.
- KV stores support local SQLite-backed create/set/get. Iroh Documents - KV stores support local SQLite-backed create/set/get. Iroh Documents
replication and command-level prefix-capability enforcement are still roadmap replication and command-level prefix-capability enforcement are still roadmap
work. The auth evaluator already understands `kv.write_prefix:<prefix>` work. The auth evaluator already understands `kv.write_prefix:<prefix>`

View file

@ -88,7 +88,7 @@ The bootstrap implementation provides:
- local filesystem CAS commands: `add`, `get`, `hash`, `has`, `pin`, `unpin`, - local filesystem CAS commands: `add`, `get`, `hash`, `has`, `pin`, `unpin`,
`cleanup`, `list` `cleanup`, `list`
- local DB resource registration: `geth db add <name> <path>` and - local DB resource registration: `geth db add <name> <path>` and
`geth db status <name>` `geth db status <name>` with schema and `crsql_changes` metadata
- local SQLite-backed KV commands: `geth kv create/set/get` - local SQLite-backed KV commands: `geth kv create/set/get`
- local JSON document commands: `geth document create/status/set/get` - local JSON document commands: `geth document create/status/set/get`
- local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub` - local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub`

View file

@ -878,6 +878,19 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> {
.unwrap_or_else(|| "unknown".to_owned()) .unwrap_or_else(|| "unknown".to_owned())
); );
println!("schema_metadata: {}", db.schema_metadata); println!("schema_metadata: {}", db.schema_metadata);
println!(
"crsqlite_changes_available: {}",
db.crsqlite_changes.available
);
if let Some(count) = db.crsqlite_changes.change_count {
println!("crsqlite_change_count: {count}");
}
if let Some(version) = db.crsqlite_changes.max_db_version {
println!("crsqlite_max_db_version: {version}");
}
if let Some(error) = db.crsqlite_changes.error {
println!("crsqlite_changes_error: {error}");
}
println!("sync_status: {}", db.sync_status); println!("sync_status: {}", db.sync_status);
} }
ControlResponse::KvCreated { kv } => { ControlResponse::KvCreated { kv } => {

View file

@ -12,9 +12,43 @@ pub struct DbResource {
pub path_exists: bool, pub path_exists: bool,
pub size_bytes: Option<u64>, pub size_bytes: Option<u64>,
pub schema_metadata: String, pub schema_metadata: String,
pub crsqlite_changes: CrSqliteChangeMetadata,
pub sync_status: String, pub sync_status: String,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CrSqliteChangeMetadata {
pub available: bool,
pub change_count: Option<u64>,
pub max_db_version: Option<i64>,
pub columns: Vec<String>,
pub error: Option<String>,
}
impl CrSqliteChangeMetadata {
#[must_use]
pub fn unavailable() -> Self {
Self {
available: false,
change_count: None,
max_db_version: None,
columns: Vec::new(),
error: None,
}
}
#[must_use]
pub fn error(error: impl Into<String>) -> Self {
Self {
available: false,
change_count: None,
max_db_version: None,
columns: Vec::new(),
error: Some(error.into()),
}
}
}
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum DbError { pub enum DbError {
#[error("invalid db resource name: {0}")] #[error("invalid db resource name: {0}")]
@ -78,6 +112,43 @@ pub fn schema_metadata(path: &Path) -> Result<String, DbError> {
)) ))
} }
pub fn crsqlite_change_metadata(path: &Path) -> Result<CrSqliteChangeMetadata, DbError> {
let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?;
let available: bool = conn.query_row(
r#"SELECT EXISTS(
SELECT 1 FROM sqlite_master
WHERE name = 'crsql_changes' AND type IN ('table', 'view')
)"#,
[],
|row| row.get(0),
)?;
if !available {
return Ok(CrSqliteChangeMetadata::unavailable());
}
let mut columns_stmt = conn.prepare("PRAGMA table_info('crsql_changes')")?;
let columns = columns_stmt
.query_map([], |row| row.get::<_, String>(1))?
.collect::<Result<Vec<_>, _>>()?;
let change_count: u64 =
conn.query_row("SELECT COUNT(*) FROM crsql_changes", [], |row| row.get(0))?;
let max_db_version = if columns.iter().any(|column| column == "db_version") {
conn.query_row("SELECT MAX(db_version) FROM crsql_changes", [], |row| {
row.get(0)
})?
} else {
None
};
Ok(CrSqliteChangeMetadata {
available: true,
change_count: Some(change_count),
max_db_version,
columns,
error: None,
})
}
#[must_use] #[must_use]
pub fn crsqlite_sync_roadmap() -> &'static str { pub fn crsqlite_sync_roadmap() -> &'static str {
"future db sync reads crsql_changes, exchanges changes over Iroh, and applies through crsql_changes" "future db sync reads crsql_changes, exchanges changes over Iroh, and applies through crsql_changes"
@ -119,4 +190,39 @@ mod tests {
assert!(first.contains("indexes=1")); assert!(first.contains("indexes=1"));
assert!(first.contains("schema_hash=")); assert!(first.contains("schema_hash="));
} }
#[test]
fn crsqlite_change_metadata_reports_mock_change_table() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("notes.sqlite");
let conn = Connection::open(&path).expect("open sqlite");
conn.execute(
r#"CREATE TABLE crsql_changes(
table_name TEXT NOT NULL,
pk TEXT NOT NULL,
cid TEXT NOT NULL,
val BLOB,
col_version INTEGER NOT NULL,
db_version INTEGER NOT NULL,
site_id BLOB,
cl INTEGER,
seq INTEGER
)"#,
[],
)
.expect("create crsql_changes table");
conn.execute(
"INSERT INTO crsql_changes(table_name, pk, cid, val, col_version, db_version) VALUES ('notes', '1', 'body', 'hello', 1, 7)",
[],
)
.expect("insert change");
drop(conn);
let metadata = crsqlite_change_metadata(&path).expect("change metadata");
assert!(metadata.available);
assert_eq!(metadata.change_count, Some(1));
assert_eq!(metadata.max_db_version, Some(7));
assert!(metadata.columns.contains(&"db_version".to_owned()));
}
} }

View file

@ -716,6 +716,7 @@ pub fn handle_request(
} }
let path = std::fs::canonicalize(path)?; let path = std::fs::canonicalize(path)?;
let schema_metadata = geth_db::schema_metadata(&path)?; let schema_metadata = geth_db::schema_metadata(&path)?;
let crsqlite_changes = geth_db::crsqlite_change_metadata(&path)?;
let resource_id = format!("resource:db:{name}"); let resource_id = format!("resource:db:{name}");
let db_id = format!("db:{name}"); let db_id = format!("db:{name}");
let resource = StoredResource { let resource = StoredResource {
@ -733,7 +734,7 @@ pub fn handle_request(
}; };
store.insert_db_resource(&stored)?; store.insert_db_resource(&stored)?;
Ok(ControlResponse::DbAdded { Ok(ControlResponse::DbAdded {
db: db_resource_from_stored_with_schema(&stored, schema_metadata), db: db_resource_from_stored_with_schema(&stored, schema_metadata, crsqlite_changes),
}) })
} }
ControlRequest::DbStatus { name } => { ControlRequest::DbStatus { name } => {
@ -910,17 +911,32 @@ fn stored_resource_to_descriptor(stored: StoredResource) -> Result<ResourceDescr
fn db_resource_from_stored(stored: &StoredDbResource) -> Result<DbResource, NodeError> { fn db_resource_from_stored(stored: &StoredDbResource) -> Result<DbResource, NodeError> {
let path = Path::new(&stored.path); let path = Path::new(&stored.path);
let metadata = path.metadata().ok(); let metadata = path.metadata().ok();
let schema_metadata = if metadata.as_ref().is_some_and(std::fs::Metadata::is_file) { let (schema_metadata, crsqlite_changes) = if metadata
geth_db::schema_metadata(path).unwrap_or_else(|error| format!("unavailable:{error}")) .as_ref()
.is_some_and(std::fs::Metadata::is_file)
{
(
geth_db::schema_metadata(path).unwrap_or_else(|error| format!("unavailable:{error}")),
geth_db::crsqlite_change_metadata(path)
.unwrap_or_else(|error| geth_db::CrSqliteChangeMetadata::error(error.to_string())),
)
} else { } else {
"unavailable:path-missing".to_owned() (
"unavailable:path-missing".to_owned(),
geth_db::CrSqliteChangeMetadata::error("path-missing"),
)
}; };
Ok(db_resource_from_stored_with_schema(stored, schema_metadata)) Ok(db_resource_from_stored_with_schema(
stored,
schema_metadata,
crsqlite_changes,
))
} }
fn db_resource_from_stored_with_schema( fn db_resource_from_stored_with_schema(
stored: &StoredDbResource, stored: &StoredDbResource,
schema_metadata: String, schema_metadata: String,
crsqlite_changes: geth_db::CrSqliteChangeMetadata,
) -> DbResource { ) -> DbResource {
let path = Path::new(&stored.path); let path = Path::new(&stored.path);
let metadata = path.metadata().ok(); let metadata = path.metadata().ok();
@ -932,6 +948,7 @@ fn db_resource_from_stored_with_schema(
path_exists: metadata.as_ref().is_some_and(std::fs::Metadata::is_file), path_exists: metadata.as_ref().is_some_and(std::fs::Metadata::is_file),
size_bytes: metadata.map(|metadata| metadata.len()), size_bytes: metadata.map(|metadata| metadata.len()),
schema_metadata, schema_metadata,
crsqlite_changes,
sync_status: "local-only".to_owned(), sync_status: "local-only".to_owned(),
} }
} }

View file

@ -399,6 +399,23 @@ fn db_add_and_status_register_local_db_metadata() {
[], [],
) )
.expect("create notes table"); .expect("create notes table");
conn.execute(
r#"CREATE TABLE crsql_changes(
table_name TEXT NOT NULL,
pk TEXT NOT NULL,
cid TEXT NOT NULL,
val BLOB,
col_version INTEGER NOT NULL,
db_version INTEGER NOT NULL
)"#,
[],
)
.expect("create crsql_changes");
conn.execute(
"INSERT INTO crsql_changes(table_name, pk, cid, val, col_version, db_version) VALUES ('notes', '1', 'body', 'hello', 1, 3)",
[],
)
.expect("insert crsql change");
drop(conn); drop(conn);
let response = geth_node::handle_request( let response = geth_node::handle_request(
@ -415,8 +432,11 @@ fn db_add_and_status_register_local_db_metadata() {
assert_eq!(db.sync_status, "local-only"); assert_eq!(db.sync_status, "local-only");
assert!(db.path_exists); assert!(db.path_exists);
assert!(db.size_bytes.unwrap_or_default() > 0); assert!(db.size_bytes.unwrap_or_default() > 0);
assert!(db.schema_metadata.contains("tables=1")); assert!(db.schema_metadata.contains("tables=2"));
assert!(db.schema_metadata.contains("schema_hash=")); assert!(db.schema_metadata.contains("schema_hash="));
assert!(db.crsqlite_changes.available);
assert_eq!(db.crsqlite_changes.change_count, Some(1));
assert_eq!(db.crsqlite_changes.max_db_version, Some(3));
} }
other => panic!("unexpected response: {other:?}"), other => panic!("unexpected response: {other:?}"),
} }
@ -432,8 +452,13 @@ fn db_add_and_status_register_local_db_metadata() {
geth_control::ControlResponse::DbStatus { db } => { geth_control::ControlResponse::DbStatus { db } => {
assert_eq!(db.name, "notes"); assert_eq!(db.name, "notes");
assert!(db.path.ends_with("notes.sqlite")); assert!(db.path.ends_with("notes.sqlite"));
assert!(db.schema_metadata.contains("tables=1")); assert!(db.schema_metadata.contains("tables=2"));
assert!(db.schema_metadata.contains("schema_hash=")); assert!(db.schema_metadata.contains("schema_hash="));
assert!(
db.crsqlite_changes
.columns
.contains(&"db_version".to_owned())
);
} }
other => panic!("unexpected response: {other:?}"), other => panic!("unexpected response: {other:?}"),
} }

View file

@ -94,8 +94,10 @@ blobs. Iroh-blobs, providers, encrypted blobs, richer cache policies, manifests,
and file sync trees are future work. and file sync trees are future work.
`geth-db` currently registers local SQLite paths as DB resources and reports `geth-db` currently registers local SQLite paths as DB resources and reports
local-only sync status plus a read-only SQLite schema summary/hash. cr-sqlite local-only sync status plus a read-only SQLite schema summary/hash. It also
loading, change extraction, and DB sync are future work. inspects `crsql_changes` metadata when that table or view exists, reporting
change count, columns, and max `db_version`. Loading cr-sqlite, extracting
change batches, applying changes, and DB sync are future work.
`geth-kv` currently provides a SQLite-backed local fallback for named KV stores `geth-kv` currently provides a SQLite-backed local fallback for named KV stores
through `kv create/set/get`. Iroh Documents namespaces, prefix authorization through `kv create/set/get`. Iroh Documents namespaces, prefix authorization

View file

@ -304,9 +304,12 @@ Automerge documents.
- `[ ]` cr-sqlite change extraction. - `[ ]` cr-sqlite change extraction.
Acceptance criteria: Acceptance criteria:
- The module can read changes from `crsql_changes`. - `[x]` DB status detects whether `crsql_changes` exists.
- Schema hash/version metadata is included in sync batches. - `[x]` DB status reports `crsql_changes` row count, columns, and max
- Tests use a temp SQLite DB and deterministic fixture changes. `db_version` when available.
- `[x]` Tests use a temp SQLite DB and deterministic fixture changes.
- `[ ]` The module can extract typed change batches from `crsql_changes`.
- `[ ]` Schema hash/version metadata is included in sync batches.
- `[ ]` DB sync over Iroh. - `[ ]` DB sync over Iroh.
Acceptance criteria: Acceptance criteria: