diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 4fd7215..ef2aee5 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -3355,7 +3355,14 @@ async fn keychain_sync_from_peer_since( invalid_ops_rejected += 1; continue; } - store_keychain_op(&store, &op)?; + let stored_signatures = valid_signatures + .iter() + .map(stored_keychain_signature_from_signature) + .collect::>(); + store.insert_keychain_op_with_signatures( + &stored_keychain_op_from_op(&op)?, + &stored_signatures, + )?; if existing.is_none() { local_ops.push(op.clone()); trusted_admins = geth_keychain::reduce_keychain_ops(&local_ops).admin_keys; @@ -3367,14 +3374,6 @@ async fn keychain_sync_from_peer_since( signature.signer.to_string(), signature.namespace.clone(), ); - store.insert_keychain_signature(&StoredKeychainSignature { - op_id: signature.op_id.to_string(), - signer: signature.signer.to_string(), - signer_public_key: signature.signer_public_key.clone(), - namespace: signature.namespace.clone(), - signature: signature.signature.clone(), - created_at_ms: signature.created_at.0, - })?; signatures_imported += usize::from(known_signatures.insert(signature_key)); } } @@ -3475,7 +3474,14 @@ async fn auth_sync_from_peer_since( invalid_ops_rejected += 1; continue; } - store_auth_op(&store, &op)?; + let stored_signatures = valid_signatures + .iter() + .map(stored_auth_signature_from_signature) + .collect::>(); + store.insert_auth_op_with_signatures( + &stored_auth_op_from_op(&op)?, + &stored_signatures, + )?; ops_imported += usize::from(existing.is_none()); for signature in valid_signatures { let signature_key = ( @@ -3483,14 +3489,6 @@ async fn auth_sync_from_peer_since( signature.signer.to_string(), signature.namespace.clone(), ); - store.insert_auth_signature(&StoredAuthSignature { - op_id: signature.op_id.to_string(), - signer: signature.signer.to_string(), - signer_public_key: signature.signer_public_key.clone(), - namespace: signature.namespace.clone(), - signature: signature.signature.clone(), - created_at_ms: signature.created_at.0, - })?; signatures_imported += usize::from(known_signatures.insert(signature_key)); } } @@ -9393,13 +9391,17 @@ fn overlay_packets( } fn store_auth_op(store: &Store, op: &AuthOp) -> Result<(), NodeError> { - store.insert_auth_op(&StoredAuthOp { + store.insert_auth_op(&stored_auth_op_from_op(op)?)?; + Ok(()) +} + +fn stored_auth_op_from_op(op: &AuthOp) -> Result { + Ok(StoredAuthOp { op_id: op.id.to_string(), resource_id: op.resource.to_string(), op_json: serde_json::to_string(op)?, created_at_ms: op.created_at.0, - })?; - Ok(()) + }) } fn store_and_sign_auth_ops( @@ -9571,12 +9573,16 @@ fn verify_auth_signature_with_ssh( } fn store_keychain_op(store: &Store, op: &KeychainOp) -> Result<(), NodeError> { - store.insert_keychain_op(&StoredKeychainOp { + store.insert_keychain_op(&stored_keychain_op_from_op(op)?)?; + Ok(()) +} + +fn stored_keychain_op_from_op(op: &KeychainOp) -> Result { + Ok(StoredKeychainOp { op_id: op.id.to_string(), op_json: serde_json::to_string(op)?, created_at_ms: op.created_at.0, - })?; - Ok(()) + }) } #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] @@ -9633,31 +9639,30 @@ fn import_verified_sigchain( ) }) .collect::>(); + let mut signatures_imported = 0; let mut ops_imported = 0; for op in ops { + let op_signatures = signatures + .iter() + .filter(|signature| signature.op_id == op.id) + .map(stored_keychain_signature_from_signature) + .collect::>(); + store + .insert_keychain_op_with_signatures(&stored_keychain_op_from_op(op)?, &op_signatures)?; if !existing_ops.contains(&op.id) { - store_keychain_op(store, op)?; ops_imported += 1; } - } - let mut signatures_imported = 0; - for signature in signatures { - let key = ( - signature.op_id.clone(), - signature.signer.clone(), - signature.namespace.clone(), - signature.signature.clone(), - ); - if !existing_signatures.contains(&key) { - store.insert_keychain_signature(&StoredKeychainSignature { - op_id: signature.op_id.to_string(), - signer: signature.signer.to_string(), - signer_public_key: signature.signer_public_key.clone(), - namespace: signature.namespace.clone(), - signature: signature.signature.clone(), - created_at_ms: signature.created_at.0, - })?; - signatures_imported += 1; + for signature in signatures + .iter() + .filter(|signature| signature.op_id == op.id) + { + let key = ( + signature.op_id.clone(), + signature.signer.clone(), + signature.namespace.clone(), + signature.signature.clone(), + ); + signatures_imported += usize::from(!existing_signatures.contains(&key)); } } Ok(geth_control::KeychainFetchImportReport { diff --git a/crates/geth-store/src/lib.rs b/crates/geth-store/src/lib.rs index b88d8f4..660bf1e 100644 --- a/crates/geth-store/src/lib.rs +++ b/crates/geth-store/src/lib.rs @@ -1001,6 +1001,36 @@ impl Store { Ok(()) } + pub fn insert_auth_op_with_signatures( + &self, + op: &StoredAuthOp, + signatures: &[StoredAuthSignature], + ) -> Result<(), StoreError> { + let tx = self.conn.unchecked_transaction()?; + tx.execute( + r#"INSERT OR REPLACE INTO auth_ops(op_id, resource_id, op_json, created_at_ms) + VALUES (?1, ?2, ?3, ?4)"#, + params![op.op_id, op.resource_id, op.op_json, op.created_at_ms], + )?; + for signature in signatures { + tx.execute( + r#"INSERT OR REPLACE INTO auth_signatures( + op_id, signer, signer_public_key, namespace, signature, created_at_ms + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)"#, + params![ + signature.op_id, + signature.signer, + signature.signer_public_key, + signature.namespace, + signature.signature, + signature.created_at_ms + ], + )?; + } + tx.commit()?; + Ok(()) + } + pub fn list_auth_ops_for_resource( &self, resource_id: &str, @@ -1084,6 +1114,36 @@ impl Store { Ok(()) } + pub fn insert_keychain_op_with_signatures( + &self, + op: &StoredKeychainOp, + signatures: &[StoredKeychainSignature], + ) -> Result<(), StoreError> { + let tx = self.conn.unchecked_transaction()?; + tx.execute( + r#"INSERT OR REPLACE INTO keychain_ops(op_id, op_json, created_at_ms) + VALUES (?1, ?2, ?3)"#, + params![op.op_id, op.op_json, op.created_at_ms], + )?; + for signature in signatures { + tx.execute( + r#"INSERT OR REPLACE INTO keychain_signatures( + op_id, signer, signer_public_key, namespace, signature, created_at_ms + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)"#, + params![ + signature.op_id, + signature.signer, + signature.signer_public_key, + signature.namespace, + signature.signature, + signature.created_at_ms + ], + )?; + } + tx.commit()?; + Ok(()) + } + pub fn list_keychain_ops(&self) -> Result, StoreError> { let mut stmt = self.conn.prepare( r#"SELECT op_id, op_json, created_at_ms @@ -1958,6 +2018,44 @@ mod tests { ); } + #[test] + fn auth_op_with_signatures_rolls_back_when_signature_insert_fails() { + let store = Store::open_memory().expect("open"); + let op = StoredAuthOp { + op_id: "auth-op:atomic".to_owned(), + resource_id: "resource:test".to_owned(), + op_json: "{}".to_owned(), + created_at_ms: 1, + }; + let signature = StoredAuthSignature { + op_id: op.op_id.clone(), + signer: "key:admin".to_owned(), + signer_public_key: "ssh-ed25519 AAAA test".to_owned(), + namespace: "geth.auth.v1@geth.local".to_owned(), + signature: b"sig".to_vec(), + created_at_ms: 2, + }; + store + .conn + .execute("DROP TABLE auth_signatures", []) + .expect("drop signatures table"); + + assert!( + store + .insert_auth_op_with_signatures(&op, &[signature]) + .is_err() + ); + let count: i64 = store + .conn + .query_row( + "SELECT COUNT(*) FROM auth_ops WHERE op_id = ?1", + [&op.op_id], + |row| row.get(0), + ) + .expect("count auth ops"); + assert_eq!(count, 0); + } + #[test] fn keychain_ops_roundtrip() { let store = Store::open_memory().expect("open"); diff --git a/docs/production-readiness-roadmap.md b/docs/production-readiness-roadmap.md index 66f40e1..819c771 100644 --- a/docs/production-readiness-roadmap.md +++ b/docs/production-readiness-roadmap.md @@ -127,12 +127,12 @@ it. - `[x]` Fresh database creation and repeated opens are idempotent. - `[x]` Old schema fixtures migrate to the current schema in tests. -- `[ ]` Make multi-table writes transactional. +- `[~]` Make multi-table writes transactional. Acceptance criteria: - - `[ ]` Signed operation and signature imports commit atomically. + - `[x]` Signed operation and signature imports commit atomically. - `[ ]` Multi-record sync imports cannot leave partial state after a local error. - - `[ ]` Tests cover failure injection for at least one multi-table path. + - `[x]` Tests cover failure injection for at least one multi-table path. - `[ ]` Add backup and restore workflow. Acceptance criteria: