authored_signed_fact_tests.rs (11404B)
1 use super::*; 2 use crate::{OpenMode, OpenOptions, Paths}; 3 use radroots_storage::{ 4 authored::WorkClaim, 5 authored_atomic::{ 6 ApplySignedArtifact, ApplyWorkFailure, CancelAuthoredWork, ClaimAuthoredWork, 7 }, 8 event::SourceGeneration, 9 }; 10 use tempfile::TempDir; 11 12 use super::signed_fact_fixture as fixture; 13 use fixture::*; 14 15 pub(super) async fn open(temp: &TempDir, mode: OpenMode) -> SqliteStorage { 16 let options = OpenOptions::new(Paths::from_directory(temp.path()).unwrap(), mode); 17 let options = if matches!(mode, OpenMode::Create) { 18 options 19 .with_source_generation(SourceGeneration::new([9; 32]).unwrap(), 9) 20 .unwrap() 21 } else { 22 options 23 }; 24 SqliteStorage::open(options).await.unwrap() 25 } 26 27 pub(super) async fn prepared( 28 temp: &TempDir, 29 ) -> (SqliteStorage, radroots_event::SignedEvent, WorkClaim) { 30 let store = open(temp, OpenMode::Create).await; 31 let (preparation, event) = prepare(); 32 store.execute_authored(preparation).await.unwrap(); 33 let artifact = store.authored_artifact(ids().1).await.unwrap().unwrap(); 34 let active = claim(artifact.revision(), 4, 11); 35 store 36 .execute_authored(AuthoredAtomicCommand::Claim(ClaimAuthoredWork::new( 37 ClaimAuthoredTarget::ArtifactSigning(ids().1), 38 active.clone(), 39 ))) 40 .await 41 .unwrap(); 42 (store, event, active) 43 } 44 45 fn fence(claim: &WorkClaim) -> WorkFence { 46 WorkFence::new(*claim.token(), claim.generation(), claim.row_revision()).unwrap() 47 } 48 49 #[tokio::test] 50 async fn late_stopped_signature_reopens_with_exact_bytes_and_no_scheduling_authority() { 51 for cancelled in [false, true] { 52 let temp = TempDir::new().unwrap(); 53 let (store, event, active) = prepared(&temp).await; 54 let artifact = store.authored_artifact(ids().1).await.unwrap().unwrap(); 55 let stop = if cancelled { 56 AuthoredAtomicCommand::Cancel( 57 CancelAuthoredWork::new( 58 CancelAuthoredTarget::ArtifactSigning(ids().1), 59 artifact.revision(), 60 20, 61 ) 62 .unwrap(), 63 ) 64 } else { 65 AuthoredAtomicCommand::ApplyFailure( 66 ApplyWorkFailure::new( 67 AuthoredWorkTarget::Artifact(ids().1), 68 fence(&active), 69 WorkFailure::new( 70 "signing_stopped", 71 WorkPhase::Signing, 72 FailureClass::Terminal, 73 None, 74 None, 75 ) 76 .unwrap(), 77 None, 78 20, 79 ) 80 .unwrap(), 81 ) 82 }; 83 let stop_receipt = store.execute_authored(stop).await.unwrap(); 84 let command = record(event, active, 40); 85 let receipt = store.execute_authored(command.clone()).await.unwrap(); 86 let retained = store.authored_artifact(ids().1).await.unwrap().unwrap(); 87 assert_eq!( 88 retained.signing_state(), 89 if cancelled { 90 SigningState::Cancelled 91 } else { 92 SigningState::FailedTerminal 93 } 94 ); 95 assert_eq!(retained.signed().unwrap().event().raw_json(), RAW); 96 let (physical, stop): (String, Option<String>) = sqlx::query_as( 97 "SELECT signing_state, signing_stop FROM radroots_runtime_authored_artifacts", 98 ) 99 .fetch_one(store.pool()) 100 .await 101 .unwrap(); 102 assert_eq!(physical, "signed"); 103 assert_eq!( 104 stop.as_deref(), 105 Some(if cancelled { 106 "cancelled" 107 } else { 108 "failed_terminal" 109 }) 110 ); 111 for mutation in [ 112 "UPDATE radroots_runtime_authored_artifacts SET signed_raw_json = x'7b7d'", 113 "UPDATE radroots_runtime_authored_artifacts SET signed_raw_sha256 = zeroblob(32)", 114 "UPDATE radroots_runtime_authored_artifacts SET signing_stop = NULL", 115 ] { 116 assert!(sqlx::query(mutation).execute(store.pool()).await.is_err()); 117 } 118 for target in [ 119 ClaimAuthoredTarget::ArtifactSigning(ids().1), 120 ClaimAuthoredTarget::ArtifactAdmission(ids().1), 121 ClaimAuthoredTarget::DeliveryPlan(ids().2), 122 ] { 123 assert!( 124 store 125 .execute_authored(AuthoredAtomicCommand::Claim(ClaimAuthoredWork::new( 126 target, 127 claim(retained.revision(), 8, 50), 128 ))) 129 .await 130 .is_err() 131 ); 132 } 133 store.close().await.unwrap(); 134 let store = open(&temp, OpenMode::ReadWriteExisting).await; 135 assert_eq!( 136 store.authored_artifact(ids().1).await.unwrap().unwrap(), 137 retained 138 ); 139 assert_eq!( 140 store 141 .authored_receipt(stop_receipt.commit_id()) 142 .await 143 .unwrap() 144 .unwrap(), 145 stop_receipt 146 ); 147 let replay = store.execute_authored(command).await.unwrap(); 148 assert_eq!(replay.disposition(), AtomicCommitDisposition::Replay); 149 assert_eq!(replay.outcome(), receipt.outcome()); 150 let delivery = store 151 .authored_delivery_plan(ids().2) 152 .await 153 .unwrap() 154 .unwrap(); 155 assert!(delivery.request().is_none()); 156 assert!(delivery.attempts().is_empty()); 157 store.close().await.unwrap(); 158 } 159 } 160 161 #[tokio::test] 162 async fn actual_commit_failure_rolls_back_late_fact_binding_and_receipt_then_retries() { 163 let temp = TempDir::new().unwrap(); 164 let (store, event, active) = prepared(&temp).await; 165 let before = store.authored_artifact(ids().1).await.unwrap().unwrap(); 166 let command = record(event, active, 40); 167 sqlx::query("CREATE TABLE signed_fact_commit_fault (parent BLOB REFERENCES radroots_runtime_authored_operations(operation_id) DEFERRABLE INITIALLY DEFERRED)") 168 .execute(store.pool()).await.unwrap(); 169 sqlx::query("CREATE TRIGGER signed_fact_commit_fault_trigger AFTER INSERT ON radroots_runtime_authored_atomic_commits WHEN NEW.phase = 'signing' BEGIN INSERT INTO signed_fact_commit_fault VALUES (x'99999999999999999999999999999999'); END") 170 .execute(store.pool()).await.unwrap(); 171 assert!(store.execute_authored(command.clone()).await.is_err()); 172 assert_eq!( 173 store.authored_artifact(ids().1).await.unwrap().unwrap(), 174 before 175 ); 176 assert!( 177 store 178 .authored_delivery_plan(ids().2) 179 .await 180 .unwrap() 181 .unwrap() 182 .request() 183 .is_none() 184 ); 185 assert!( 186 store 187 .authored_receipt(command.commit_id()) 188 .await 189 .unwrap() 190 .is_none() 191 ); 192 assert_eq!( 193 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM signed_fact_commit_fault") 194 .fetch_one(store.pool()) 195 .await 196 .unwrap(), 197 0 198 ); 199 sqlx::query("DROP TRIGGER signed_fact_commit_fault_trigger") 200 .execute(store.pool()) 201 .await 202 .unwrap(); 203 sqlx::query("DROP TABLE signed_fact_commit_fault") 204 .execute(store.pool()) 205 .await 206 .unwrap(); 207 store.close().await.unwrap(); 208 let store = open(&temp, OpenMode::ReadWriteExisting).await; 209 assert_eq!( 210 store.authored_artifact(ids().1).await.unwrap().unwrap(), 211 before 212 ); 213 let receipt = store.execute_authored(command.clone()).await.unwrap(); 214 assert_eq!(receipt.disposition(), AtomicCommitDisposition::Committed); 215 assert_eq!( 216 store 217 .authored_delivery_plan(ids().2) 218 .await 219 .unwrap() 220 .unwrap() 221 .request() 222 .unwrap() 223 .payload() 224 .event() 225 .raw_json(), 226 RAW 227 ); 228 store.close().await.unwrap(); 229 let store = open(&temp, OpenMode::ReadOnly).await; 230 assert_eq!( 231 store 232 .authored_receipt(command.commit_id()) 233 .await 234 .unwrap() 235 .unwrap(), 236 receipt 237 ); 238 assert!(store.execute_authored(command).await.is_err()); 239 store.close().await.unwrap(); 240 } 241 242 #[tokio::test] 243 async fn stale_active_fence_and_altered_provenance_cannot_install_late_facts() { 244 let temp = TempDir::new().unwrap(); 245 let (store, event, active) = prepared(&temp).await; 246 let before = store.authored_artifact(ids().1).await.unwrap().unwrap(); 247 assert!( 248 store 249 .execute_authored(AuthoredAtomicCommand::ApplySigned( 250 ApplySignedArtifact::new(ids().1, fence(&active), event.clone(), 40,).unwrap() 251 )) 252 .await 253 .is_err() 254 ); 255 let wrong_owner = WorkClaim::new( 256 *active.token(), 257 "wrong-owner", 258 active.generation(), 259 11, 260 31, 261 active.row_revision(), 262 ) 263 .unwrap(); 264 for command in [ 265 record(event.clone(), wrong_owner, 40), 266 record(fixture::event(OTHER_RAW), active.clone(), 40), 267 ] { 268 assert!(store.execute_authored(command.clone()).await.is_err()); 269 assert!( 270 store 271 .authored_receipt(command.commit_id()) 272 .await 273 .unwrap() 274 .is_none() 275 ); 276 assert_eq!( 277 store.authored_artifact(ids().1).await.unwrap().unwrap(), 278 before 279 ); 280 } 281 let command = record(event, active.clone(), 40); 282 store.execute_authored(command).await.unwrap(); 283 let retained = store.authored_artifact(ids().1).await.unwrap().unwrap(); 284 let alternate = record(fixture::event(&format!(" {RAW} ")), active, 50); 285 assert_eq!( 286 store.execute_authored(alternate.clone()).await, 287 Err(Error::AtomicCommitConflict) 288 ); 289 assert!( 290 store 291 .authored_receipt(alternate.commit_id()) 292 .await 293 .unwrap() 294 .is_none() 295 ); 296 assert_eq!( 297 store.authored_artifact(ids().1).await.unwrap().unwrap(), 298 retained 299 ); 300 store.close().await.unwrap(); 301 } 302 303 #[tokio::test] 304 async fn failure_while_binding_delivery_rolls_back_the_first_signed_fact() { 305 let temp = TempDir::new().unwrap(); 306 let (store, event, active) = prepared(&temp).await; 307 let before = store.authored_artifact(ids().1).await.unwrap().unwrap(); 308 let command = record(event, active, 40); 309 sqlx::query("CREATE TRIGGER signed_fact_binding_fault BEFORE UPDATE ON radroots_runtime_authored_delivery_plans BEGIN SELECT RAISE(ABORT, 'fixture delivery binding failure'); END") 310 .execute(store.pool()).await.unwrap(); 311 assert!(store.execute_authored(command.clone()).await.is_err()); 312 assert_eq!( 313 store.authored_artifact(ids().1).await.unwrap().unwrap(), 314 before 315 ); 316 assert!( 317 store 318 .authored_receipt(command.commit_id()) 319 .await 320 .unwrap() 321 .is_none() 322 ); 323 sqlx::query("DROP TRIGGER signed_fact_binding_fault") 324 .execute(store.pool()) 325 .await 326 .unwrap(); 327 store.execute_authored(command).await.unwrap(); 328 store.close().await.unwrap(); 329 }