authored_delivery_reconciliation_tests.rs (23022B)
1 use super::signed_fact_fixture::{claim, ids, record}; 2 use super::*; 3 use crate::OpenMode; 4 use radroots_storage::{ 5 authored::WorkClaim, 6 authored_atomic::{ClaimAuthoredWork, ReconcileDeliveryFacts}, 7 }; 8 use tempfile::TempDir; 9 10 async fn signed(temp: &TempDir) -> SqliteStorage { 11 let (store, event, signing) = signed_fact_tests::prepared(temp).await; 12 store 13 .execute_authored(record(event, signing, 12)) 14 .await 15 .unwrap(); 16 store 17 } 18 19 async fn plan(store: &SqliteStorage) -> AuthoredDeliveryPlan { 20 store 21 .authored_delivery_plan(ids().2) 22 .await 23 .unwrap() 24 .unwrap() 25 } 26 27 async fn issue(store: &SqliteStorage) -> (WorkClaim, AuthoredAtomicReceipt) { 28 let active = claim(plan(store).await.revision(), 5, 13); 29 let receipt = store 30 .execute_authored(AuthoredAtomicCommand::Claim(ClaimAuthoredWork::new( 31 ClaimAuthoredTarget::DeliveryPlan(ids().2), 32 active.clone(), 33 ))) 34 .await 35 .unwrap(); 36 (active, receipt) 37 } 38 39 fn command(plan: &AuthoredDeliveryPlan) -> AuthoredAtomicCommand { 40 AuthoredAtomicCommand::ReconcileDelivery( 41 ReconcileDeliveryFacts::new(plan, None, None, 51).unwrap(), 42 ) 43 } 44 45 #[tokio::test] 46 async fn claim_and_reconciliation_provenance_reopen_exactly_and_survive_stop() { 47 let temp = TempDir::new().unwrap(); 48 let store = signed(&temp).await; 49 assert!( 50 store 51 .authored_delivery_history(ids().2) 52 .await 53 .unwrap() 54 .unwrap() 55 .proves_no_issued_attempt() 56 ); 57 let (active, issued) = issue(&store).await; 58 let history = store 59 .authored_delivery_history(ids().2) 60 .await 61 .unwrap() 62 .unwrap(); 63 assert_eq!(history.claims().len(), 1); 64 assert!(history.has_unresolved_claims()); 65 store 66 .execute_authored(delivery_fact_tests::fact( 67 &plan(&store).await, 68 active.clone(), 69 )) 70 .await 71 .unwrap(); 72 let before = plan(&store).await; 73 let command = command(&before); 74 let committed = store.execute_authored(command.clone()).await.unwrap(); 75 let after = plan(&store).await; 76 assert_eq!(after.state(), AuthoredDeliveryState::Satisfied); 77 assert_eq!(after.attempt_count(), 1); 78 assert_eq!(after.attempts()[0].claim_evidence(), Some(&active)); 79 assert_eq!(after.delivery_facts(), before.delivery_facts()); 80 assert_eq!( 81 store 82 .authored_receipt(issued.commit_id()) 83 .await 84 .unwrap() 85 .unwrap(), 86 issued 87 ); 88 assert!( 89 !store 90 .authored_delivery_history(ids().2) 91 .await 92 .unwrap() 93 .unwrap() 94 .has_unresolved_claims() 95 ); 96 for sql in [ 97 "DELETE FROM radroots_runtime_authored_delivery_claims", 98 "UPDATE radroots_runtime_authored_delivery_claims SET claim_id = claim_id", 99 "DELETE FROM radroots_runtime_authored_delivery_reconciliations", 100 "UPDATE radroots_runtime_authored_delivery_reconciliations SET attempt = 2", 101 ] { 102 assert!(sqlx::query(sql).execute(store.pool()).await.is_err()); 103 } 104 store.close().await.unwrap(); 105 let store = signed_fact_tests::open(&temp, OpenMode::ReadWriteExisting).await; 106 assert_eq!(plan(&store).await, after); 107 let replay = store.execute_authored(command).await.unwrap(); 108 assert_eq!(replay.disposition(), AtomicCommitDisposition::Replay); 109 assert_eq!(replay.outcome(), committed.outcome()); 110 store 111 .execute_authored(AuthoredAtomicCommand::Cancel( 112 radroots_storage::authored_atomic::CancelAuthoredWork::new( 113 CancelAuthoredTarget::DeliveryPlan(ids().2), 114 after.revision(), 115 60, 116 ) 117 .unwrap(), 118 )) 119 .await 120 .unwrap(); 121 let stopped = plan(&store).await; 122 assert_eq!(stopped.stop_requested_at_unix_ms(), Some(60)); 123 assert_eq!(stopped.attempts(), after.attempts()); 124 assert_eq!(stopped.delivery_facts(), after.delivery_facts()); 125 store.close().await.unwrap(); 126 } 127 128 #[tokio::test] 129 async fn reconciliation_commit_failure_rolls_back_marker_plan_and_receipt() { 130 let temp = TempDir::new().unwrap(); 131 let store = signed(&temp).await; 132 let (active, _) = issue(&store).await; 133 store 134 .execute_authored(delivery_fact_tests::fact(&plan(&store).await, active)) 135 .await 136 .unwrap(); 137 let before = plan(&store).await; 138 let before_history = store 139 .authored_delivery_history(ids().2) 140 .await 141 .unwrap() 142 .unwrap(); 143 let command = command(&before); 144 sqlx::query("CREATE TABLE reconciliation_commit_fault (parent BLOB REFERENCES radroots_runtime_authored_operations(operation_id) DEFERRABLE INITIALLY DEFERRED)").execute(store.pool()).await.unwrap(); 145 sqlx::query("CREATE TRIGGER reconciliation_commit_fault_trigger AFTER INSERT ON radroots_runtime_authored_delivery_reconciliations BEGIN INSERT INTO reconciliation_commit_fault VALUES (x'99999999999999999999999999999999'); END").execute(store.pool()).await.unwrap(); 146 assert!(store.execute_authored(command.clone()).await.is_err()); 147 assert_eq!(plan(&store).await, before); 148 assert_eq!( 149 store 150 .authored_delivery_history(ids().2) 151 .await 152 .unwrap() 153 .unwrap(), 154 before_history 155 ); 156 assert!( 157 store 158 .authored_receipt(command.commit_id()) 159 .await 160 .unwrap() 161 .is_none() 162 ); 163 assert_eq!( 164 sqlx::query_scalar::<_, i64>( 165 "SELECT COUNT(*) FROM radroots_runtime_authored_delivery_reconciliations" 166 ) 167 .fetch_one(store.pool()) 168 .await 169 .unwrap(), 170 0 171 ); 172 sqlx::query("DROP TRIGGER reconciliation_commit_fault_trigger") 173 .execute(store.pool()) 174 .await 175 .unwrap(); 176 store.execute_authored(command).await.unwrap(); 177 assert_eq!(plan(&store).await.attempt_count(), 1); 178 // The existing writer may replace these rows only within its transaction; 179 // a committed missing attempt must never orphan retained reconciliation. 180 assert!( 181 sqlx::query("DELETE FROM radroots_runtime_authored_delivery_attempts") 182 .execute(store.pool()) 183 .await 184 .is_err() 185 ); 186 assert_eq!(plan(&store).await.attempt_count(), 1); 187 store.close().await.unwrap(); 188 } 189 190 #[tokio::test] 191 async fn original_claim_commit_failure_preserves_no_issued_attempt_proof() { 192 let temp = TempDir::new().unwrap(); 193 let store = signed(&temp).await; 194 let before = plan(&store).await; 195 let active = claim(before.revision(), 5, 13); 196 let command = AuthoredAtomicCommand::Claim(ClaimAuthoredWork::new( 197 ClaimAuthoredTarget::DeliveryPlan(ids().2), 198 active, 199 )); 200 sqlx::query("CREATE TABLE claim_index_commit_fault (parent BLOB REFERENCES radroots_runtime_authored_operations(operation_id) DEFERRABLE INITIALLY DEFERRED)").execute(store.pool()).await.unwrap(); 201 sqlx::query("CREATE TRIGGER claim_index_commit_fault_trigger AFTER INSERT ON radroots_runtime_authored_delivery_claims BEGIN INSERT INTO claim_index_commit_fault VALUES (x'99999999999999999999999999999999'); END").execute(store.pool()).await.unwrap(); 202 assert!(store.execute_authored(command.clone()).await.is_err()); 203 assert_eq!(plan(&store).await, before); 204 assert!( 205 store 206 .authored_receipt(command.commit_id()) 207 .await 208 .unwrap() 209 .is_none() 210 ); 211 assert!( 212 store 213 .authored_delivery_history(ids().2) 214 .await 215 .unwrap() 216 .unwrap() 217 .proves_no_issued_attempt() 218 ); 219 sqlx::query("DROP TRIGGER claim_index_commit_fault_trigger") 220 .execute(store.pool()) 221 .await 222 .unwrap(); 223 store.execute_authored(command).await.unwrap(); 224 let issued = store 225 .authored_delivery_history(ids().2) 226 .await 227 .unwrap() 228 .unwrap(); 229 assert_eq!(issued.claims().len(), 1); 230 assert!(issued.has_unresolved_claims()); 231 assert!(!issued.proves_no_issued_attempt()); 232 store.close().await.unwrap(); 233 } 234 235 #[tokio::test] 236 async fn history_read_snapshot_cannot_mix_old_plan_with_new_claim_or_reconciliation() { 237 let temp = TempDir::new().unwrap(); 238 let store = signed(&temp).await; 239 let mut read = store.pool().begin().await.unwrap(); 240 let frozen = delivery_reconciliation::history(&mut read, ids().2) 241 .await 242 .unwrap() 243 .unwrap(); 244 assert!(frozen.proves_no_issued_attempt()); 245 let (active, _) = issue(&store).await; 246 assert_eq!( 247 delivery_reconciliation::history(&mut read, ids().2) 248 .await 249 .unwrap() 250 .unwrap(), 251 frozen 252 ); 253 read.rollback().await.unwrap(); 254 let mut read = store.pool().begin().await.unwrap(); 255 let issued = delivery_reconciliation::history(&mut read, ids().2) 256 .await 257 .unwrap() 258 .unwrap(); 259 store 260 .execute_authored(delivery_fact_tests::fact(&plan(&store).await, active)) 261 .await 262 .unwrap(); 263 store 264 .execute_authored(command(&plan(&store).await)) 265 .await 266 .unwrap(); 267 assert_eq!( 268 delivery_reconciliation::history(&mut read, ids().2) 269 .await 270 .unwrap() 271 .unwrap(), 272 issued 273 ); 274 read.rollback().await.unwrap(); 275 let after = store 276 .authored_delivery_history(ids().2) 277 .await 278 .unwrap() 279 .unwrap(); 280 assert!(!after.has_unresolved_claims()); 281 assert_eq!(after.plan().attempt_count(), 1); 282 let mut stale = store.pool().begin().await.unwrap(); 283 assert_eq!( 284 delivery_reconciliation::persist(&mut stale, issued.plan()).await, 285 Err(Error::InvalidAuthoredDeliveryPlan) 286 ); 287 stale.rollback().await.unwrap(); 288 assert_eq!( 289 store 290 .authored_delivery_history(ids().2) 291 .await 292 .unwrap() 293 .unwrap(), 294 after 295 ); 296 store.close().await.unwrap(); 297 } 298 299 #[tokio::test] 300 async fn oversized_historical_claims_remain_retained_but_cannot_authorize_new_work() { 301 let temp = TempDir::new().unwrap(); 302 let store = signed(&temp).await; 303 let mut current = plan(&store).await; 304 let mut transaction = store.pool().begin().await.unwrap(); 305 // Historical databases may exceed the new admission bound. Construct their 306 // typed original receipts without invoking the now-bounded command path. 307 for index in 0..1025u64 { 308 let at = 13 + index * 21; 309 let active = WorkClaim::new( 310 [7; 16], 311 "historical-worker", 312 std::num::NonZeroU64::new(index + 5).unwrap(), 313 at, 314 at + 20, 315 current.revision(), 316 ) 317 .unwrap(); 318 current.claim(active.clone(), at).unwrap(); 319 let command = AuthoredAtomicCommand::Claim(ClaimAuthoredWork::new( 320 ClaimAuthoredTarget::DeliveryPlan(ids().2), 321 active, 322 )); 323 let receipt = AuthoredAtomicReceipt::new( 324 &command, 325 AtomicCommitDisposition::Committed, 326 at, 327 AuthoredAtomicOutcome::DeliveryPlan(current.clone()), 328 ) 329 .unwrap(); 330 sqlx::query("INSERT INTO radroots_runtime_authored_atomic_commits (commit_id, commit_digest, phase, target_id, requested_at_unix_ms, committed_at_unix_ms, receipt) VALUES (?, ?, 'claim', ?, ?, ?, ?)") 331 .bind(receipt.commit_id().as_bytes().as_slice()).bind(receipt.digest().as_bytes().as_slice()).bind(ids().2.as_bytes().as_slice()).bind(at as i64).bind(at as i64) 332 .bind(serde_json::to_vec(&serde_json::json!({"outcome": receipt.outcome()})).unwrap()).execute(&mut *transaction).await.unwrap(); 333 sqlx::query("INSERT INTO radroots_runtime_authored_delivery_claims (plan_id, claim_id) VALUES (?, ?)") 334 .bind(ids().2.as_bytes().as_slice()).bind(receipt.commit_id().as_bytes().as_slice()).execute(&mut *transaction).await.unwrap(); 335 } 336 persist_plan(&mut transaction, ¤t).await.unwrap(); 337 transaction.commit().await.unwrap(); 338 let history = store 339 .authored_delivery_history(ids().2) 340 .await 341 .unwrap() 342 .unwrap(); 343 assert_eq!(history.claims().len(), 1024); 344 assert!(history.is_truncated()); 345 assert!(!history.is_complete()); 346 assert!(!history.proves_no_issued_attempt()); 347 assert!(history.has_unresolved_claims()); 348 assert_eq!( 349 history.require_pending_fact_provenance(), 350 Err(Error::DeliveryAttemptOverflow) 351 ); 352 let at = 13 + 1025 * 21; 353 let active = WorkClaim::new( 354 [8; 16], 355 "new-worker", 356 std::num::NonZeroU64::new(2048).unwrap(), 357 at, 358 at + 20, 359 current.revision(), 360 ) 361 .unwrap(); 362 let command = AuthoredAtomicCommand::Claim(ClaimAuthoredWork::new( 363 ClaimAuthoredTarget::DeliveryPlan(ids().2), 364 active, 365 )); 366 assert_eq!( 367 store.execute_authored(command.clone()).await, 368 Err(Error::DeliveryAttemptOverflow) 369 ); 370 assert_eq!(plan(&store).await, current); 371 assert!( 372 store 373 .authored_receipt(command.commit_id()) 374 .await 375 .unwrap() 376 .is_none() 377 ); 378 assert_eq!( 379 sqlx::query_scalar::<_, i64>( 380 "SELECT COUNT(*) FROM radroots_runtime_authored_delivery_claims" 381 ) 382 .fetch_one(store.pool()) 383 .await 384 .unwrap(), 385 1025 386 ); 387 store.close().await.unwrap(); 388 } 389 390 #[tokio::test] 391 async fn forged_indexed_claim_receipt_fails_closed_instead_of_proving_absence() { 392 let temp = TempDir::new().unwrap(); 393 let store = signed(&temp).await; 394 let (_, issued) = issue(&store).await; 395 let before = plan(&store).await; 396 sqlx::query("INSERT INTO radroots_runtime_authored_atomic_commits (commit_id, commit_digest, phase, target_id, requested_at_unix_ms, committed_at_unix_ms, receipt) VALUES (?, ?, 'claim', ?, 13, 13, ?)") 397 .bind([9u8; 16].as_slice()).bind([8u8; 32].as_slice()).bind(ids().2.as_bytes().as_slice()) 398 .bind(serde_json::to_vec(&serde_json::json!({"outcome": issued.outcome()})).unwrap()).execute(store.pool()).await.unwrap(); 399 sqlx::query( 400 "INSERT INTO radroots_runtime_authored_delivery_claims (plan_id, claim_id) VALUES (?, ?)", 401 ) 402 .bind(ids().2.as_bytes().as_slice()) 403 .bind([9u8; 16].as_slice()) 404 .execute(store.pool()) 405 .await 406 .unwrap(); 407 assert!(store.authored_delivery_history(ids().2).await.is_err()); 408 assert_eq!(plan(&store).await, before); 409 store.close().await.unwrap(); 410 } 411 412 #[tokio::test] 413 async fn late_legacy_marker_can_precede_existing_marker_without_rewriting_it() { 414 use radroots_storage::{ 415 authored::{FailureClass, RetrySchedule, WorkFailure, WorkPhase}, 416 authored_atomic::{ApplyDeliveryAttempt, RecordDeliveryFact, WorkFence}, 417 authored_delivery::DeliveryAttemptOutcome, 418 }; 419 use radroots_transport::{ 420 DeliveryReceipt, outcome::DeliveryOutcome, sink::DeliveryTargetReceipt, 421 }; 422 let retry = |attempt, at| { 423 RetrySchedule::new( 424 std::num::NonZeroU32::new(attempt).unwrap(), 425 at, 426 WorkFailure::new( 427 "delivery_pending", 428 WorkPhase::Delivery, 429 FailureClass::Retryable, 430 Some(at), 431 None, 432 ) 433 .unwrap(), 434 ) 435 .unwrap() 436 }; 437 let fence = |claim: &WorkClaim| { 438 WorkFence::new(*claim.token(), claim.generation(), claim.row_revision()).unwrap() 439 }; 440 let temp = TempDir::new().unwrap(); 441 let store = signed(&temp).await; 442 let (first, _) = issue(&store).await; 443 let current = plan(&store).await; 444 let request = current.request().unwrap(); 445 let outcome = DeliveryAttemptOutcome::Receipt( 446 DeliveryReceipt::for_request( 447 request, 448 request 449 .target_set() 450 .targets() 451 .iter() 452 .cloned() 453 .map(|target| { 454 DeliveryTargetReceipt::attempted(target, DeliveryOutcome::unavailable()) 455 }) 456 .collect(), 457 ) 458 .unwrap(), 459 ); 460 store 461 .execute_authored(AuthoredAtomicCommand::ApplyDelivery( 462 ApplyDeliveryAttempt::new( 463 ids().2, 464 fence(&first), 465 outcome.clone(), 466 Some(retry(1, 18)), 467 14, 468 ) 469 .unwrap(), 470 )) 471 .await 472 .unwrap(); 473 let second = claim(plan(&store).await.revision(), 6, 20); 474 store 475 .execute_authored(AuthoredAtomicCommand::Claim(ClaimAuthoredWork::new( 476 ClaimAuthoredTarget::DeliveryPlan(ids().2), 477 second.clone(), 478 ))) 479 .await 480 .unwrap(); 481 store 482 .execute_authored(AuthoredAtomicCommand::RecordDelivery( 483 RecordDeliveryFact::new(ids().2, ids().1, second.clone(), outcome.clone(), 21).unwrap(), 484 )) 485 .await 486 .unwrap(); 487 store 488 .execute_authored(AuthoredAtomicCommand::ReconcileDelivery( 489 ReconcileDeliveryFacts::new( 490 &plan(&store).await, 491 Some(fence(&second)), 492 Some(retry(2, 25)), 493 22, 494 ) 495 .unwrap(), 496 )) 497 .await 498 .unwrap(); 499 let marker_before: Vec<u8> = sqlx::query_scalar( 500 "SELECT claim_id FROM radroots_runtime_authored_delivery_reconciliations WHERE attempt = 2", 501 ) 502 .fetch_one(store.pool()) 503 .await 504 .unwrap(); 505 store 506 .execute_authored(AuthoredAtomicCommand::RecordDelivery( 507 RecordDeliveryFact::new(ids().2, ids().1, first.clone(), outcome, 50).unwrap(), 508 )) 509 .await 510 .unwrap(); 511 store 512 .execute_authored(AuthoredAtomicCommand::ReconcileDelivery( 513 ReconcileDeliveryFacts::new(&plan(&store).await, None, Some(retry(2, 60)), 51).unwrap(), 514 )) 515 .await 516 .unwrap(); 517 let after = plan(&store).await; 518 assert_eq!(after.attempt_count(), 2); 519 assert_eq!(after.attempts()[0].recorded_at_unix_ms(), 14); 520 assert_eq!(after.attempts()[0].claim_evidence(), Some(&first)); 521 assert_eq!(after.attempts()[1].recorded_at_unix_ms(), 22); 522 assert_eq!(after.attempts()[1].claim_evidence(), Some(&second)); 523 assert_eq!(sqlx::query_scalar::<_, Vec<u8>>("SELECT claim_id FROM radroots_runtime_authored_delivery_reconciliations WHERE attempt = 2").fetch_one(store.pool()).await.unwrap(), marker_before); 524 store.close().await.unwrap(); 525 let store = signed_fact_tests::open(&temp, OpenMode::ReadWriteExisting).await; 526 assert_eq!(plan(&store).await, after); 527 assert!( 528 !store 529 .authored_delivery_history(ids().2) 530 .await 531 .unwrap() 532 .unwrap() 533 .has_unresolved_claims() 534 ); 535 store.close().await.unwrap(); 536 } 537 538 #[tokio::test] 539 async fn history_rejects_ambiguous_preparation_and_distinguishes_missing_plan() { 540 let temp = TempDir::new().unwrap(); 541 let store = signed(&temp).await; 542 assert!( 543 store 544 .authored_delivery_history(AuthoredDeliveryPlanId::new([99; 16]).unwrap()) 545 .await 546 .unwrap() 547 .is_none() 548 ); 549 let before = plan(&store).await; 550 sqlx::query("INSERT INTO radroots_runtime_authored_atomic_commits (commit_id, commit_digest, phase, target_id, requested_at_unix_ms, committed_at_unix_ms, receipt) SELECT ?, commit_digest, phase, target_id, requested_at_unix_ms, committed_at_unix_ms, receipt FROM radroots_runtime_authored_atomic_commits WHERE phase = 'prepare'") 551 .bind([9u8; 16].as_slice()).execute(store.pool()).await.unwrap(); 552 assert_eq!( 553 store.authored_delivery_history(ids().2).await, 554 Err(Error::AtomicWorkflowMismatch) 555 ); 556 assert_eq!(plan(&store).await, before); 557 store.close().await.unwrap(); 558 } 559 560 #[tokio::test] 561 async fn missing_or_rebound_normalized_marker_cannot_return_a_valid_plan() { 562 for missing in [false, true] { 563 let temp = TempDir::new().unwrap(); 564 let store = signed(&temp).await; 565 let (first, _) = issue(&store).await; 566 let second = claim(plan(&store).await.revision(), 6, 40); 567 let second_command = AuthoredAtomicCommand::Claim(ClaimAuthoredWork::new( 568 ClaimAuthoredTarget::DeliveryPlan(ids().2), 569 second, 570 )); 571 store 572 .execute_authored(second_command.clone()) 573 .await 574 .unwrap(); 575 store 576 .execute_authored(delivery_fact_tests::fact(&plan(&store).await, first)) 577 .await 578 .unwrap(); 579 store 580 .execute_authored(AuthoredAtomicCommand::ReconcileDelivery( 581 ReconcileDeliveryFacts::new(&plan(&store).await, None, None, 61).unwrap(), 582 )) 583 .await 584 .unwrap(); 585 let before: Vec<u8> = 586 sqlx::query_scalar("SELECT snapshot FROM radroots_runtime_authored_delivery_plans") 587 .fetch_one(store.pool()) 588 .await 589 .unwrap(); 590 // Isolated corruption fixtures deliberately remove the applicable guard. 591 // The normal command path cannot erase or rebind these immutable rows. 592 if missing { 593 sqlx::query( 594 "DROP TRIGGER radroots_runtime_authored_delivery_reconciliations_delete_guard", 595 ) 596 .execute(store.pool()) 597 .await 598 .unwrap(); 599 sqlx::query("DELETE FROM radroots_runtime_authored_delivery_reconciliations") 600 .execute(store.pool()) 601 .await 602 .unwrap(); 603 } else { 604 sqlx::query( 605 "DROP TRIGGER radroots_runtime_authored_delivery_reconciliations_update_guard", 606 ) 607 .execute(store.pool()) 608 .await 609 .unwrap(); 610 sqlx::query( 611 "UPDATE radroots_runtime_authored_delivery_reconciliations SET claim_id = ?", 612 ) 613 .bind(second_command.commit_id().as_bytes().as_slice()) 614 .execute(store.pool()) 615 .await 616 .unwrap(); 617 } 618 assert_eq!( 619 store.authored_delivery_plan(ids().2).await, 620 Err(Error::InvalidAuthoredDeliveryPlan) 621 ); 622 assert_eq!( 623 store.authored_delivery_history(ids().2).await, 624 Err(Error::InvalidAuthoredDeliveryPlan) 625 ); 626 assert_eq!( 627 sqlx::query_scalar::<_, Vec<u8>>( 628 "SELECT snapshot FROM radroots_runtime_authored_delivery_plans" 629 ) 630 .fetch_one(store.pool()) 631 .await 632 .unwrap(), 633 before 634 ); 635 store.close().await.unwrap(); 636 } 637 }