diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 5e24b70..5c1e482 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -3328,6 +3328,7 @@ async fn keychain_sync_from_peer_since( .into_iter() .map(|signature| (signature.op_id, signature.signer, signature.namespace)) .collect::>(); + let mut accepted_imports = Vec::new(); let mut ops_imported = 0; let mut signatures_imported = 0; let mut invalid_ops_rejected = 0; @@ -3358,10 +3359,8 @@ async fn keychain_sync_from_peer_since( invalid_ops_rejected += 1; continue; } - let existing = load_keychain_ops(&store)? - .into_iter() - .find(|existing| existing.id == op.id); - if existing.as_ref().is_some_and(|existing| existing != &op) { + let existing = local_ops.iter().find(|existing| existing.id == op.id); + if existing.as_ref().is_some_and(|existing| *existing != &op) { invalid_ops_rejected += 1; continue; } @@ -3369,10 +3368,6 @@ async fn keychain_sync_from_peer_since( .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; @@ -3386,7 +3381,9 @@ async fn keychain_sync_from_peer_since( ); signatures_imported += usize::from(known_signatures.insert(signature_key)); } + accepted_imports.push((stored_keychain_op_from_op(&op)?, stored_signatures)); } + store.insert_keychain_ops_with_signatures(&accepted_imports)?; store_live_sync_cursor(&store, peer_node, "keychain", high_water_ms)?; Ok(ControlResponse::KeychainSynced { peer_node_id: node_id, @@ -3447,6 +3444,8 @@ async fn auth_sync_from_peer_since( .into_iter() .map(|signature| (signature.op_id, signature.signer, signature.namespace)) .collect::>(); + let mut local_ops = load_auth_ops(&store)?; + let mut accepted_imports = Vec::new(); let mut ops_imported = 0; let mut signatures_imported = 0; let mut invalid_ops_rejected = 0; @@ -3477,10 +3476,8 @@ async fn auth_sync_from_peer_since( invalid_ops_rejected += 1; continue; } - let existing = load_auth_ops(&store)? - .into_iter() - .find(|existing| existing.id == op.id); - if existing.as_ref().is_some_and(|existing| existing != &op) { + let existing = local_ops.iter().find(|existing| existing.id == op.id); + if existing.as_ref().is_some_and(|existing| *existing != &op) { invalid_ops_rejected += 1; continue; } @@ -3488,11 +3485,10 @@ async fn auth_sync_from_peer_since( .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()); + if existing.is_none() { + local_ops.push(op.clone()); + ops_imported += 1; + } for signature in valid_signatures { let signature_key = ( signature.op_id.to_string(), @@ -3501,7 +3497,9 @@ async fn auth_sync_from_peer_since( ); signatures_imported += usize::from(known_signatures.insert(signature_key)); } + accepted_imports.push((stored_auth_op_from_op(&op)?, stored_signatures)); } + store.insert_auth_ops_with_signatures(&accepted_imports)?; store_live_sync_cursor(&store, peer_node, "auth", high_water_ms)?; Ok(ControlResponse::AuthSynced { peer_node_id: node_id, diff --git a/crates/geth-store/src/lib.rs b/crates/geth-store/src/lib.rs index 767df97..47413c2 100644 --- a/crates/geth-store/src/lib.rs +++ b/crates/geth-store/src/lib.rs @@ -1052,25 +1052,18 @@ impl Store { 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 - ], - )?; + Self::insert_auth_op_with_signatures_tx(&tx, op, signatures)?; + tx.commit()?; + Ok(()) + } + + pub fn insert_auth_ops_with_signatures( + &self, + ops: &[(StoredAuthOp, Vec)], + ) -> Result<(), StoreError> { + let tx = self.conn.unchecked_transaction()?; + for (op, signatures) in ops { + Self::insert_auth_op_with_signatures_tx(&tx, op, signatures)?; } tx.commit()?; Ok(()) @@ -1165,25 +1158,18 @@ impl Store { 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 - ], - )?; + Self::insert_keychain_op_with_signatures_tx(&tx, op, signatures)?; + tx.commit()?; + Ok(()) + } + + pub fn insert_keychain_ops_with_signatures( + &self, + ops: &[(StoredKeychainOp, Vec)], + ) -> Result<(), StoreError> { + let tx = self.conn.unchecked_transaction()?; + for (op, signatures) in ops { + Self::insert_keychain_op_with_signatures_tx(&tx, op, signatures)?; } tx.commit()?; Ok(()) @@ -1245,6 +1231,62 @@ impl Store { .map_err(StoreError::from) } + fn insert_auth_op_with_signatures_tx( + tx: &Transaction<'_>, + op: &StoredAuthOp, + signatures: &[StoredAuthSignature], + ) -> Result<(), StoreError> { + 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 + ], + )?; + } + Ok(()) + } + + fn insert_keychain_op_with_signatures_tx( + tx: &Transaction<'_>, + op: &StoredKeychainOp, + signatures: &[StoredKeychainSignature], + ) -> Result<(), StoreError> { + 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 + ], + )?; + } + Ok(()) + } + pub fn insert_node_enrollment_request( &self, request: &StoredNodeEnrollmentRequest, @@ -2120,6 +2162,52 @@ mod tests { assert_eq!(count, 0); } + #[test] + fn auth_ops_with_signatures_batch_commits_multiple_ops() { + let store = Store::open_memory().expect("open"); + let first = StoredAuthOp { + op_id: "auth-op:batch:1".to_owned(), + resource_id: "resource:test".to_owned(), + op_json: r#"{"id":"auth-op:batch:1"}"#.to_owned(), + created_at_ms: 1, + }; + let second = StoredAuthOp { + op_id: "auth-op:batch:2".to_owned(), + resource_id: "resource:test".to_owned(), + op_json: r#"{"id":"auth-op:batch:2"}"#.to_owned(), + created_at_ms: 2, + }; + let signature = |op: &StoredAuthOp| 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: op.created_at_ms, + }; + + store + .insert_auth_ops_with_signatures(&[ + (first.clone(), vec![signature(&first)]), + (second.clone(), vec![signature(&second)]), + ]) + .expect("batch insert auth ops"); + + assert_eq!( + store + .list_auth_ops_for_resource("resource:test") + .expect("list auth ops"), + vec![first, second] + ); + assert_eq!( + store + .list_auth_signatures() + .expect("list auth signatures") + .len(), + 2 + ); + } + #[test] fn keychain_ops_roundtrip() { let store = Store::open_memory().expect("open"); @@ -2143,6 +2231,48 @@ mod tests { ); } + #[test] + fn keychain_ops_with_signatures_batch_commits_multiple_ops() { + let store = Store::open_memory().expect("open"); + let first = StoredKeychainOp { + op_id: "keychain-op:batch:1".to_owned(), + op_json: r#"{"id":"keychain-op:batch:1"}"#.to_owned(), + created_at_ms: 1, + }; + let second = StoredKeychainOp { + op_id: "keychain-op:batch:2".to_owned(), + op_json: r#"{"id":"keychain-op:batch:2"}"#.to_owned(), + created_at_ms: 2, + }; + let signature = |op: &StoredKeychainOp| StoredKeychainSignature { + op_id: op.op_id.clone(), + signer: "key:admin".to_owned(), + signer_public_key: "ssh-ed25519 AAAA test".to_owned(), + namespace: "geth.keychain.v1@geth.local".to_owned(), + signature: b"sig".to_vec(), + created_at_ms: op.created_at_ms, + }; + + store + .insert_keychain_ops_with_signatures(&[ + (first.clone(), vec![signature(&first)]), + (second.clone(), vec![signature(&second)]), + ]) + .expect("batch insert keychain ops"); + + assert_eq!( + store.list_keychain_ops().expect("list keychain ops"), + vec![first, second] + ); + assert_eq!( + store + .list_keychain_signatures() + .expect("list keychain signatures") + .len(), + 2 + ); + } + #[test] fn keychain_signatures_roundtrip() { let store = Store::open_memory().expect("open"); diff --git a/docs/production-readiness-roadmap.md b/docs/production-readiness-roadmap.md index f78570a..e872118 100644 --- a/docs/production-readiness-roadmap.md +++ b/docs/production-readiness-roadmap.md @@ -130,6 +130,8 @@ it. - `[~]` Make multi-table writes transactional. Acceptance criteria: - `[x]` Signed operation and signature imports commit atomically. + - `[x]` Keychain/auth sync imports stage accepted records and commit accepted + op/signature groups through store batch transactions. - `[ ]` Multi-record sync imports cannot leave partial state after a local error. - `[x]` Tests cover failure injection for at least one multi-table path.