authored_signed_durability_tests.rs (5862B)
1 use super::{ 2 tests::{fence, ids, prepare, signed}, 3 *, 4 }; 5 use crate::{OpenMode, OpenOptions, Paths}; 6 use core::num::NonZeroU64; 7 use radroots_storage::{ 8 authored::WorkClaim, 9 authored_atomic::{ApplySignedArtifact, ClaimAuthoredWork}, 10 event::SourceGeneration, 11 }; 12 use tempfile::TempDir; 13 14 async fn open(temp: &TempDir, mode: OpenMode) -> SqliteStorage { 15 let options = OpenOptions::new(Paths::from_directory(temp.path()).unwrap(), mode); 16 let options = if mode == OpenMode::Create { 17 options 18 .with_source_generation(SourceGeneration::new([9; 32]).unwrap(), 9) 19 .unwrap() 20 } else { 21 options 22 }; 23 SqliteStorage::open(options).await.unwrap() 24 } 25 26 #[tokio::test] 27 async fn signed_artifact_commit_failure_rolls_back_and_retries_exactly_after_reopen() { 28 let temp = TempDir::new().unwrap(); 29 let store = open(&temp, OpenMode::Create).await; 30 let (preparation, plan) = prepare(); 31 store.execute_authored(preparation).await.unwrap(); 32 let initial = store.authored_artifact(ids().1).await.unwrap().unwrap(); 33 let claim = WorkClaim::new( 34 [4; 16], 35 "sqlite-signer", 36 NonZeroU64::MIN, 37 11, 38 50, 39 initial.revision(), 40 ) 41 .unwrap(); 42 store 43 .execute_authored(AuthoredAtomicCommand::Claim(ClaimAuthoredWork::new( 44 ClaimAuthoredTarget::ArtifactSigning(ids().1), 45 claim.clone(), 46 ))) 47 .await 48 .unwrap(); 49 let artifact_before = store.authored_artifact(ids().1).await.unwrap().unwrap(); 50 let delivery_before = store 51 .authored_delivery_plan(ids().2) 52 .await 53 .unwrap() 54 .unwrap(); 55 let operation_before = store.authored_operation(ids().0).await.unwrap().unwrap(); 56 assert!(artifact_before.signed().is_none()); 57 assert!(delivery_before.request().is_none()); 58 59 // This storage fixture has a verified event ID. Cryptographic signature 60 // verification belongs to signing/Sync, not this transaction boundary. 61 let event = signed(&plan); 62 let command = AuthoredAtomicCommand::ApplySigned( 63 ApplySignedArtifact::new(ids().1, fence(&claim), event.clone(), 12).unwrap(), 64 ); 65 sqlx::query("CREATE TABLE signed_commit_fault (parent BLOB REFERENCES radroots_runtime_authored_operations(operation_id) DEFERRABLE INITIALLY DEFERRED)") 66 .execute(store.pool()).await.unwrap(); 67 // Every statement, including the final receipt INSERT, succeeds. Only 68 // SQLite's actual COMMIT rejects the deferred foreign-key violation. 69 sqlx::query("CREATE TRIGGER signed_commit_fault_trigger AFTER INSERT ON radroots_runtime_authored_atomic_commits BEGIN INSERT INTO signed_commit_fault VALUES (x'99999999999999999999999999999999'); END") 70 .execute(store.pool()).await.unwrap(); 71 assert!(store.execute_authored(command.clone()).await.is_err()); 72 assert!( 73 store 74 .authored_receipt(command.commit_id()) 75 .await 76 .unwrap() 77 .is_none() 78 ); 79 assert_eq!( 80 store.authored_artifact(ids().1).await.unwrap().unwrap(), 81 artifact_before 82 ); 83 assert_eq!( 84 store 85 .authored_delivery_plan(ids().2) 86 .await 87 .unwrap() 88 .unwrap(), 89 delivery_before 90 ); 91 assert_eq!( 92 store.authored_operation(ids().0).await.unwrap().unwrap(), 93 operation_before 94 ); 95 let failed_writes: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM signed_commit_fault") 96 .fetch_one(store.pool()) 97 .await 98 .unwrap(); 99 assert_eq!(failed_writes, 0); 100 sqlx::query("DROP TRIGGER signed_commit_fault_trigger") 101 .execute(store.pool()) 102 .await 103 .unwrap(); 104 sqlx::query("DROP TABLE signed_commit_fault") 105 .execute(store.pool()) 106 .await 107 .unwrap(); 108 store.close().await.unwrap(); 109 110 let store = open(&temp, OpenMode::ReadWriteExisting).await; 111 assert!( 112 store 113 .authored_receipt(command.commit_id()) 114 .await 115 .unwrap() 116 .is_none() 117 ); 118 assert_eq!( 119 store.authored_artifact(ids().1).await.unwrap().unwrap(), 120 artifact_before 121 ); 122 assert_eq!( 123 store 124 .authored_delivery_plan(ids().2) 125 .await 126 .unwrap() 127 .unwrap(), 128 delivery_before 129 ); 130 assert_eq!( 131 store.authored_operation(ids().0).await.unwrap().unwrap(), 132 operation_before 133 ); 134 let receipt = store.execute_authored(command.clone()).await.unwrap(); 135 assert_eq!(receipt.disposition(), AtomicCommitDisposition::Committed); 136 store.close().await.unwrap(); 137 138 let store = open(&temp, OpenMode::ReadWriteExisting).await; 139 let artifact = store.authored_artifact(ids().1).await.unwrap().unwrap(); 140 let delivery = store 141 .authored_delivery_plan(ids().2) 142 .await 143 .unwrap() 144 .unwrap(); 145 assert_eq!(artifact.signing_state(), SigningState::Signed); 146 assert_eq!( 147 artifact.signed().unwrap().event().raw_json(), 148 event.raw_json() 149 ); 150 assert_eq!( 151 delivery.request().unwrap().payload().event().raw_json(), 152 event.raw_json() 153 ); 154 assert_eq!( 155 store 156 .authored_receipt(command.commit_id()) 157 .await 158 .unwrap() 159 .unwrap() 160 .outcome(), 161 receipt.outcome() 162 ); 163 assert_eq!( 164 store.execute_authored(command).await.unwrap().disposition(), 165 AtomicCommitDisposition::Replay 166 ); 167 assert_eq!( 168 store.authored_artifact(ids().1).await.unwrap().unwrap(), 169 artifact 170 ); 171 assert_eq!( 172 store 173 .authored_delivery_plan(ids().2) 174 .await 175 .unwrap() 176 .unwrap(), 177 delivery 178 ); 179 store.close().await.unwrap(); 180 }