fix: batch signed log sync imports

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::<BTreeSet<_>>();
+            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)?
+                let existing = local_ops.iter().find(|existing| existing.id == op.id);
-                    .into_iter()
+                if existing.as_ref().is_some_and(|existing| *existing != &op) {
-                    .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::<Vec<_>>();
-                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::<BTreeSet<_>>();
+            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)?
+                let existing = local_ops.iter().find(|existing| existing.id == op.id);
-                    .into_iter()
+                if existing.as_ref().is_some_and(|existing| *existing != &op) {
-                    .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::<Vec<_>>();
-                store.insert_auth_op_with_signatures(
+                if existing.is_none() {
-                    &stored_auth_op_from_op(&op)?,
+                    local_ops.push(op.clone());
-                    &stored_signatures,
+                    ops_imported += 1;
-                )?;
+                }
-                ops_imported += usize::from(existing.is_none());
                 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,

crates/geth-store/src/lib.rs

@@ -1052,25 +1052,18 @@ impl Store {
         signatures: &[StoredAuthSignature],
     ) -> Result<(), StoreError> {
         let tx = self.conn.unchecked_transaction()?;
-        tx.execute(
+        Self::insert_auth_op_with_signatures_tx(&tx, op, signatures)?;
-            r#"INSERT OR REPLACE INTO auth_ops(op_id, resource_id, op_json, created_at_ms)
+        tx.commit()?;
-               VALUES (?1, ?2, ?3, ?4)"#,
+        Ok(())
-            params![op.op_id, op.resource_id, op.op_json, op.created_at_ms],
+    }
-        )?;
+
-        for signature in signatures {
+    pub fn insert_auth_ops_with_signatures(
-            tx.execute(
+        &self,
-                r#"INSERT OR REPLACE INTO auth_signatures(
+        ops: &[(StoredAuthOp, Vec<StoredAuthSignature>)],
-                    op_id, signer, signer_public_key, namespace, signature, created_at_ms
+    ) -> Result<(), StoreError> {
-                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)"#,
+        let tx = self.conn.unchecked_transaction()?;
-                params![
+        for (op, signatures) in ops {
-                    signature.op_id,
+            Self::insert_auth_op_with_signatures_tx(&tx, op, signatures)?;
-                    signature.signer,
-                    signature.signer_public_key,
-                    signature.namespace,
-                    signature.signature,
-                    signature.created_at_ms
-                ],
-            )?;
         }
         tx.commit()?;
         Ok(())
@@ -1165,25 +1158,18 @@ impl Store {
         signatures: &[StoredKeychainSignature],
     ) -> Result<(), StoreError> {
         let tx = self.conn.unchecked_transaction()?;
-        tx.execute(
+        Self::insert_keychain_op_with_signatures_tx(&tx, op, signatures)?;
-            r#"INSERT OR REPLACE INTO keychain_ops(op_id, op_json, created_at_ms)
+        tx.commit()?;
-               VALUES (?1, ?2, ?3)"#,
+        Ok(())
-            params![op.op_id, op.op_json, op.created_at_ms],
+    }
-        )?;
+
-        for signature in signatures {
+    pub fn insert_keychain_ops_with_signatures(
-            tx.execute(
+        &self,
-                r#"INSERT OR REPLACE INTO keychain_signatures(
+        ops: &[(StoredKeychainOp, Vec<StoredKeychainSignature>)],
-                    op_id, signer, signer_public_key, namespace, signature, created_at_ms
+    ) -> Result<(), StoreError> {
-                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)"#,
+        let tx = self.conn.unchecked_transaction()?;
-                params![
+        for (op, signatures) in ops {
-                    signature.op_id,
+            Self::insert_keychain_op_with_signatures_tx(&tx, op, signatures)?;
-                    signature.signer,
-                    signature.signer_public_key,
-                    signature.namespace,
-                    signature.signature,
-                    signature.created_at_ms
-                ],
-            )?;
         }
         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");

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.