feat: commit signed op imports atomically

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::<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;
@@ -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::<Vec<_>>();
+                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<StoredAuthOp, NodeError> {
+    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<StoredKeychainOp, NodeError> {
+    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::<BTreeSet<_>>();
+    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::<Vec<_>>();
+        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;
         }
-    }
+        for signature in signatures
-    let mut signatures_imported = 0;
+            .iter()
-    for signature in signatures {
+            .filter(|signature| signature.op_id == op.id)
+        {
             let key = (
                 signature.op_id.clone(),
                 signature.signer.clone(),
                 signature.namespace.clone(),
                 signature.signature.clone(),
             );
-        if !existing_signatures.contains(&key) {
+            signatures_imported += usize::from(!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;
         }
     }
     Ok(geth_control::KeychainFetchImportReport {

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<Vec<StoredKeychainOp>, 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");

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: