authored_draft.rs (22751B)
1 use crate::SqliteStorage; 2 use crate::backend::map_backend; 3 use radroots_storage::{ 4 Error, 5 authored_draft::{ 6 AUTHORED_DRAFT_QUERY_LIMIT_MAX, AuthoredDraft, AuthoredDraftId, AuthoredDraftRevision, 7 AuthoredDraftStage, AuthoredDraftStore, DraftAppendDisposition, DraftAppendReceipt, 8 }, 9 event::BoxFuture, 10 }; 11 use sqlx::{Row, sqlite::SqliteRow}; 12 13 const SNAPSHOT_MAX_BYTES: usize = 16 * 1024 * 1024; 14 15 #[path = "authored_draft_query.rs"] 16 mod query; 17 18 #[path = "authored_draft_row.rs"] 19 mod bounded_row; 20 21 impl AuthoredDraftStore for SqliteStorage { 22 fn append_authored_draft_pair( 23 &self, 24 pair: radroots_storage::authored_draft_pair::AuthoredDraftPair, 25 ) -> BoxFuture<'_, Result<[DraftAppendReceipt; 2], Error>> { 26 Box::pin(async move { 27 if self.event_mode() == radroots_storage::status::EventStoreMode::ReadOnly { 28 return Err(Error::BackendUnavailable); 29 } 30 let mut transaction = self 31 .pool() 32 .begin_with("BEGIN IMMEDIATE") 33 .await 34 .map_err(map_backend)?; 35 let [first, second] = pair.drafts(); 36 let [first_expected, second_expected] = *pair.expected_heads(); 37 let a = append_tx(&mut transaction, first, first_expected).await?; 38 let b = append_tx(&mut transaction, second, second_expected).await?; 39 if a != b { 40 return Err(Error::DraftRevisionConflict); 41 } 42 if a == DraftAppendDisposition::Replay { 43 transaction.rollback().await.map_err(map_backend)?; 44 } else { 45 transaction.commit().await.map_err(map_backend)?; 46 } 47 Ok([ 48 DraftAppendReceipt::new(first.clone(), a), 49 DraftAppendReceipt::new(second.clone(), b), 50 ]) 51 }) 52 } 53 54 fn query_authored_drafts( 55 &self, 56 query: radroots_storage::authored_draft_query::AuthoredDraftQuery, 57 ) -> BoxFuture<'_, Result<radroots_storage::authored_draft_query::AuthoredDraftPage, Error>> 58 { 59 Box::pin(async move { query::page(self, query).await }) 60 } 61 62 fn append_authored_draft( 63 &self, 64 draft: AuthoredDraft, 65 expected_head: Option<AuthoredDraftRevision>, 66 ) -> BoxFuture<'_, Result<DraftAppendReceipt, Error>> { 67 Box::pin(async move { 68 draft.validate()?; 69 if self.event_mode() == radroots_storage::status::EventStoreMode::ReadOnly { 70 return Err(Error::BackendUnavailable); 71 } 72 let mut transaction = self 73 .pool() 74 .begin_with("BEGIN IMMEDIATE") 75 .await 76 .map_err(map_backend)?; 77 let disposition = append_tx(&mut transaction, &draft, expected_head).await?; 78 if disposition == DraftAppendDisposition::Replay { 79 transaction.rollback().await.map_err(map_backend)?; 80 } else { 81 transaction.commit().await.map_err(map_backend)?; 82 } 83 Ok(DraftAppendReceipt::new(draft, disposition)) 84 }) 85 } 86 87 fn authored_draft_head( 88 &self, 89 draft_id: AuthoredDraftId, 90 ) -> BoxFuture<'_, Result<Option<AuthoredDraft>, Error>> { 91 Box::pin(async move { 92 bounded_row::load(self.pool(), draft_id.as_bytes(), None) 93 .await? 94 .as_ref() 95 .map(decode_row) 96 .transpose() 97 }) 98 } 99 100 fn authored_draft_revision( 101 &self, 102 draft_id: AuthoredDraftId, 103 revision: AuthoredDraftRevision, 104 ) -> BoxFuture<'_, Result<Option<AuthoredDraft>, Error>> { 105 Box::pin(async move { 106 bounded_row::load(self.pool(), draft_id.as_bytes(), Some(revision)) 107 .await? 108 .as_ref() 109 .map(decode_row) 110 .transpose() 111 }) 112 } 113 114 fn authored_draft_heads( 115 &self, 116 author: [u8; 32], 117 limit: u16, 118 ) -> BoxFuture<'_, Result<Vec<AuthoredDraft>, Error>> { 119 Box::pin(async move { 120 if author.iter().all(|byte| *byte == 0) 121 || limit == 0 122 || limit > AUTHORED_DRAFT_QUERY_LIMIT_MAX 123 { 124 return Err(Error::InvalidAuthoredDraft); 125 } 126 sqlx::query( 127 "SELECT revisions.* 128 FROM radroots_runtime_authored_draft_revisions AS revisions 129 WHERE revisions.author = ? 130 AND revisions.revision = ( 131 SELECT MAX(head.revision) 132 FROM radroots_runtime_authored_draft_revisions AS head 133 WHERE head.draft_id = revisions.draft_id 134 ) 135 ORDER BY revisions.updated_at_unix_ms DESC, revisions.draft_id 136 LIMIT ?", 137 ) 138 .bind(author.as_slice()) 139 .bind(i64::from(limit)) 140 .fetch_all(self.pool()) 141 .await 142 .map_err(map_backend)? 143 .iter() 144 .map(decode_row) 145 .collect() 146 }) 147 } 148 } 149 150 fn encode_snapshot(draft: &AuthoredDraft) -> Result<Vec<u8>, Error> { 151 let snapshot = serde_json::to_vec(draft).map_err(|_| Error::InvalidAuthoredDraft)?; 152 if snapshot.is_empty() || snapshot.len() > SNAPSHOT_MAX_BYTES { 153 return Err(Error::InvalidAuthoredDraft); 154 } 155 Ok(snapshot) 156 } 157 158 fn decode_row(row: &SqliteRow) -> Result<AuthoredDraft, Error> { 159 let snapshot = row 160 .try_get::<Vec<u8>, _>("snapshot") 161 .map_err(|_| Error::CorruptAuthoredDraft)?; 162 if snapshot.is_empty() || snapshot.len() > SNAPSHOT_MAX_BYTES { 163 return Err(Error::CorruptAuthoredDraft); 164 } 165 let draft = serde_json::from_slice::<AuthoredDraft>(snapshot.as_slice()) 166 .map_err(|_| Error::CorruptAuthoredDraft)?; 167 let draft_id = fixed::<16>(row, "draft_id")?; 168 let revision = u64_from_i64( 169 row.try_get::<i64, _>("revision") 170 .map_err(|_| Error::CorruptAuthoredDraft)?, 171 )?; 172 let author = fixed::<32>(row, "author")?; 173 let stage = row 174 .try_get::<i64, _>("stage") 175 .map_err(|_| Error::CorruptAuthoredDraft)?; 176 let operation_id = row 177 .try_get::<Option<Vec<u8>>, _>("operation_id") 178 .map_err(|_| Error::CorruptAuthoredDraft)?; 179 let payload_sha256 = fixed::<32>(row, "payload_sha256")?; 180 let created = u64_from_i64( 181 row.try_get::<i64, _>("created_at_unix_ms") 182 .map_err(|_| Error::CorruptAuthoredDraft)?, 183 )?; 184 let updated = u64_from_i64( 185 row.try_get::<i64, _>("updated_at_unix_ms") 186 .map_err(|_| Error::CorruptAuthoredDraft)?, 187 )?; 188 let operation_matches = match (operation_id, draft.operation_id()) { 189 (None, None) => true, 190 (Some(raw), Some(expected)) => raw.as_slice() == expected.as_bytes(), 191 _ => false, 192 }; 193 let schema = row 194 .try_get::<String, _>("payload_schema") 195 .map_err(|_| Error::CorruptAuthoredDraft)?; 196 let scope = row 197 .try_get::<Option<Vec<u8>>, _>("payload_scope") 198 .map_err(|_| Error::CorruptAuthoredDraft)?; 199 if schema != draft.payload_schema() 200 || scope != draft.scope().map(|scope| scope.as_bytes().to_vec()) 201 { 202 return Err(Error::CorruptAuthoredDraft); 203 } 204 if draft.draft_id().as_bytes() != &draft_id 205 || draft.revision().get() != revision 206 || draft.author() != &author 207 || stage_code(draft.stage()) != stage 208 || !operation_matches 209 || draft.payload_sha256() != &payload_sha256 210 || draft.created_at_unix_ms() != created 211 || draft.updated_at_unix_ms() != updated 212 { 213 return Err(Error::CorruptAuthoredDraft); 214 } 215 Ok(draft) 216 } 217 218 fn fixed<const N: usize>(row: &SqliteRow, column: &str) -> Result<[u8; N], Error> { 219 row.try_get::<Vec<u8>, _>(column) 220 .map_err(|_| Error::CorruptAuthoredDraft)? 221 .try_into() 222 .map_err(|_| Error::CorruptAuthoredDraft) 223 } 224 225 const fn stage_code(stage: AuthoredDraftStage) -> i64 { 226 match stage { 227 AuthoredDraftStage::Draft => 0, 228 AuthoredDraftStage::MediaPreparing => 1, 229 AuthoredDraftStage::MediaUploading => 2, 230 AuthoredDraftStage::ReadyToSign => 3, 231 AuthoredDraftStage::Queued => 4, 232 AuthoredDraftStage::Cancelled => 5, 233 } 234 } 235 236 fn i64_from_u64(value: u64) -> Result<i64, Error> { 237 i64::try_from(value).map_err(|_| Error::InvalidAuthoredDraft) 238 } 239 240 fn u64_from_i64(value: i64) -> Result<u64, Error> { 241 u64::try_from(value).map_err(|_| Error::CorruptAuthoredDraft) 242 } 243 244 #[cfg(test)] 245 mod tests { 246 use super::*; 247 use crate::{OpenMode, OpenOptions, Paths}; 248 use radroots_storage::event::SourceGeneration; 249 use sha2::Digest; 250 use tempfile::TempDir; 251 252 fn draft(id: u8, at: u64) -> AuthoredDraft { 253 AuthoredDraft::initial( 254 AuthoredDraftId::new([id; 16]).unwrap(), 255 [7; 32], 256 "radroots.phase1-draft.v1", 257 vec![id], 258 AuthoredDraftStage::Draft, 259 None, 260 at, 261 ) 262 .unwrap() 263 } 264 265 pub(super) async fn open_store(temp: &TempDir) -> SqliteStorage { 266 let paths = Paths::from_directory(temp.path()).unwrap(); 267 SqliteStorage::open( 268 OpenOptions::new(paths, OpenMode::Create) 269 .with_source_generation(SourceGeneration::new([9; 32]).unwrap(), 9) 270 .unwrap(), 271 ) 272 .await 273 .unwrap() 274 } 275 276 #[tokio::test] 277 async fn revisions_survive_reopen_and_replay_exactly() { 278 let temp = TempDir::new().unwrap(); 279 let store = open_store(&temp).await; 280 let first = draft(1, 10); 281 assert_eq!( 282 store 283 .append_authored_draft(first.clone(), None) 284 .await 285 .unwrap() 286 .disposition(), 287 DraftAppendDisposition::Inserted 288 ); 289 assert_eq!( 290 store 291 .append_authored_draft(first.clone(), None) 292 .await 293 .unwrap() 294 .disposition(), 295 DraftAppendDisposition::Replay 296 ); 297 let second = first 298 .successor( 299 b"next".to_vec(), 300 AuthoredDraftStage::MediaPreparing, 301 None, 302 11, 303 ) 304 .unwrap(); 305 store 306 .append_authored_draft(second.clone(), Some(first.revision())) 307 .await 308 .unwrap(); 309 drop(store); 310 let reopened = open_store(&temp).await; 311 assert_eq!( 312 reopened 313 .authored_draft_head(first.draft_id()) 314 .await 315 .unwrap(), 316 Some(second.clone()) 317 ); 318 assert_eq!( 319 reopened 320 .authored_draft_revision(first.draft_id(), first.revision()) 321 .await 322 .unwrap(), 323 Some(first) 324 ); 325 assert_eq!( 326 reopened.authored_draft_heads([7; 32], 10).await.unwrap(), 327 vec![second] 328 ); 329 } 330 331 #[tokio::test] 332 async fn conflicts_and_query_bounds_fail_closed() { 333 let temp = TempDir::new().unwrap(); 334 let store = open_store(&temp).await; 335 let first = draft(1, 10); 336 store 337 .append_authored_draft(first.clone(), None) 338 .await 339 .unwrap(); 340 let conflicting = AuthoredDraft::initial( 341 first.draft_id(), 342 [7; 32], 343 first.payload_schema(), 344 b"conflict".to_vec(), 345 AuthoredDraftStage::Draft, 346 None, 347 10, 348 ) 349 .unwrap(); 350 assert_eq!( 351 store.append_authored_draft(conflicting, None).await, 352 Err(Error::DraftRevisionConflict) 353 ); 354 assert_eq!( 355 store.authored_draft_heads([7; 32], 0).await, 356 Err(Error::InvalidAuthoredDraft) 357 ); 358 assert_eq!( 359 store.authored_draft_heads([0; 32], 1).await, 360 Err(Error::InvalidAuthoredDraft) 361 ); 362 assert_eq!( 363 store 364 .authored_draft_heads([7; 32], AUTHORED_DRAFT_QUERY_LIMIT_MAX + 1) 365 .await, 366 Err(Error::InvalidAuthoredDraft) 367 ); 368 369 let successor = first 370 .successor( 371 b"next".to_vec(), 372 AuthoredDraftStage::MediaPreparing, 373 None, 374 11, 375 ) 376 .unwrap(); 377 assert_eq!( 378 store.append_authored_draft(successor.clone(), None).await, 379 Err(Error::DraftRevisionConflict) 380 ); 381 assert_eq!( 382 store 383 .append_authored_draft(successor, Some(AuthoredDraftRevision::new(2).unwrap()),) 384 .await, 385 Err(Error::DraftRevisionConflict) 386 ); 387 assert_eq!( 388 store 389 .append_authored_draft(draft(2, 20), Some(AuthoredDraftRevision::INITIAL)) 390 .await, 391 Err(Error::DraftRevisionConflict) 392 ); 393 let noninitial_first = AuthoredDraft::reconstruct( 394 AuthoredDraftId::new([3; 16]).unwrap(), 395 AuthoredDraftRevision::new(2).unwrap(), 396 [7; 32], 397 "radroots.phase1-draft.v1", 398 vec![3], 399 sha2::Sha256::digest([3]).into(), 400 AuthoredDraftStage::Draft, 401 None, 402 30, 403 30, 404 ) 405 .unwrap(); 406 assert_eq!( 407 store.append_authored_draft(noninitial_first, None).await, 408 Err(Error::DraftRevisionConflict) 409 ); 410 } 411 412 #[tokio::test] 413 async fn simultaneous_first_append_has_one_insert_and_one_exact_replay() { 414 let temp = TempDir::new().unwrap(); 415 let store = open_store(&temp).await; 416 let first = draft(2, 20); 417 let (left, right) = tokio::join!( 418 store.append_authored_draft(first.clone(), None), 419 store.append_authored_draft(first, None), 420 ); 421 let dispositions = [left.unwrap().disposition(), right.unwrap().disposition()]; 422 assert!(dispositions.contains(&DraftAppendDisposition::Inserted)); 423 assert!(dispositions.contains(&DraftAppendDisposition::Replay)); 424 } 425 426 #[tokio::test] 427 async fn read_only_append_and_u64_overflow_fail_before_mutation() { 428 let temp = TempDir::new().unwrap(); 429 let store = open_store(&temp).await; 430 let paths = Paths::from_directory(temp.path()).unwrap(); 431 drop(store); 432 let read_only = SqliteStorage::open(OpenOptions::new(paths, OpenMode::ReadOnly)) 433 .await 434 .unwrap(); 435 assert_eq!( 436 read_only.append_authored_draft(draft(3, 30), None).await, 437 Err(Error::BackendUnavailable) 438 ); 439 drop(read_only); 440 441 let store = open_store(&temp).await; 442 let overflow = AuthoredDraft::initial( 443 AuthoredDraftId::new([4; 16]).unwrap(), 444 [7; 32], 445 "radroots.phase1-draft.v1", 446 vec![4], 447 AuthoredDraftStage::Draft, 448 None, 449 i64::MAX as u64 + 1, 450 ) 451 .unwrap(); 452 assert_eq!( 453 store.append_authored_draft(overflow, None).await, 454 Err(Error::InvalidAuthoredDraft) 455 ); 456 } 457 458 #[allow(clippy::too_many_arguments)] 459 async fn insert_raw_and_decode( 460 store: &SqliteStorage, 461 draft: &AuthoredDraft, 462 draft_id: [u8; 16], 463 revision: i64, 464 author: [u8; 32], 465 stage: i64, 466 operation_id: Option<Vec<u8>>, 467 payload_sha256: [u8; 32], 468 created_at_unix_ms: i64, 469 updated_at_unix_ms: i64, 470 ) -> Result<AuthoredDraft, Error> { 471 sqlx::query( 472 "INSERT INTO radroots_runtime_authored_draft_revisions ( 473 draft_id, revision, author, stage, operation_id, payload_sha256, 474 created_at_unix_ms, updated_at_unix_ms, snapshot, payload_schema, payload_scope 475 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", 476 ) 477 .bind(draft_id.as_slice()) 478 .bind(revision) 479 .bind(author.as_slice()) 480 .bind(stage) 481 .bind(operation_id) 482 .bind(payload_sha256.as_slice()) 483 .bind(created_at_unix_ms) 484 .bind(updated_at_unix_ms) 485 .bind(encode_snapshot(draft).unwrap()) 486 .bind(draft.payload_schema()) 487 .bind(draft.scope().map(|scope| scope.as_bytes().to_vec())) 488 .execute(store.pool()) 489 .await 490 .unwrap(); 491 let row = sqlx::query( 492 "SELECT * FROM radroots_runtime_authored_draft_revisions 493 WHERE draft_id = ? AND revision = ?", 494 ) 495 .bind(draft_id.as_slice()) 496 .bind(revision) 497 .fetch_one(store.pool()) 498 .await 499 .unwrap(); 500 decode_row(&row) 501 } 502 503 #[tokio::test] 504 async fn every_redundant_draft_column_is_verified_against_the_snapshot() { 505 let temp = TempDir::new().unwrap(); 506 let store = open_store(&temp).await; 507 508 let value = draft(10, 100); 509 assert_eq!( 510 insert_raw_and_decode( 511 &store, 512 &value, 513 [99; 16], 514 1, 515 *value.author(), 516 stage_code(value.stage()), 517 None, 518 *value.payload_sha256(), 519 100, 520 100, 521 ) 522 .await, 523 Err(Error::CorruptAuthoredDraft) 524 ); 525 526 for (id, revision, author, stage, operation_id, payload_sha256, created, updated) in [ 527 ( 528 11, 529 2, 530 [7; 32], 531 0, 532 None, 533 *draft(11, 100).payload_sha256(), 534 100, 535 100, 536 ), 537 ( 538 12, 539 1, 540 [8; 32], 541 0, 542 None, 543 *draft(12, 100).payload_sha256(), 544 100, 545 100, 546 ), 547 ( 548 13, 549 1, 550 [7; 32], 551 1, 552 None, 553 *draft(13, 100).payload_sha256(), 554 100, 555 100, 556 ), 557 ( 558 14, 559 1, 560 [7; 32], 561 0, 562 Some(vec![1; 16]), 563 *draft(14, 100).payload_sha256(), 564 100, 565 100, 566 ), 567 (15, 1, [7; 32], 0, None, [9; 32], 100, 100), 568 ( 569 16, 570 1, 571 [7; 32], 572 0, 573 None, 574 *draft(16, 100).payload_sha256(), 575 99, 576 100, 577 ), 578 ( 579 17, 580 1, 581 [7; 32], 582 0, 583 None, 584 *draft(17, 100).payload_sha256(), 585 100, 586 101, 587 ), 588 ] { 589 let value = draft(id, 100); 590 assert_eq!( 591 insert_raw_and_decode( 592 &store, 593 &value, 594 [id; 16], 595 revision, 596 author, 597 stage, 598 operation_id, 599 payload_sha256, 600 created, 601 updated, 602 ) 603 .await, 604 Err(Error::CorruptAuthoredDraft), 605 "redundant column case {id}" 606 ); 607 } 608 } 609 } 610 611 #[cfg(test)] 612 #[cfg_attr(coverage_nightly, coverage(off))] 613 #[path = "authored_draft_query_tests.rs"] 614 mod query_tests; 615 616 #[cfg(test)] 617 #[cfg_attr(coverage_nightly, coverage(off))] 618 #[path = "authored_draft_all_schemas_tests.rs"] 619 mod all_schema_query_tests; 620 621 #[cfg(test)] 622 #[cfg_attr(coverage_nightly, coverage(off))] 623 #[path = "authored_durability_tests.rs"] 624 mod durability_tests; 625 626 pub(crate) async fn load_head_tx( 627 transaction: &mut sqlx::Transaction<'_, sqlx::Sqlite>, 628 id: AuthoredDraftId, 629 ) -> Result<Option<AuthoredDraft>, Error> { 630 bounded_row::load(&mut **transaction, id.as_bytes(), None) 631 .await? 632 .as_ref() 633 .map(decode_row) 634 .transpose() 635 } 636 pub(crate) async fn insert_draft_tx( 637 transaction: &mut sqlx::Transaction<'_, sqlx::Sqlite>, 638 draft: &AuthoredDraft, 639 ) -> Result<(), Error> { 640 let snapshot = encode_snapshot(draft)?; 641 sqlx::query( 642 "INSERT INTO radroots_runtime_authored_draft_revisions ( 643 draft_id, revision, author, stage, operation_id, payload_sha256, 644 created_at_unix_ms, updated_at_unix_ms, snapshot, payload_schema, payload_scope 645 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", 646 ) 647 .bind(draft.draft_id().as_bytes().as_slice()) 648 .bind(i64_from_u64(draft.revision().get())?) 649 .bind(draft.author().as_slice()) 650 .bind(stage_code(draft.stage())) 651 .bind(draft.operation_id().map(|id| id.as_bytes().to_vec())) 652 .bind(draft.payload_sha256().as_slice()) 653 .bind(i64_from_u64(draft.created_at_unix_ms())?) 654 .bind(i64_from_u64(draft.updated_at_unix_ms())?) 655 .bind(snapshot) 656 .bind(draft.payload_schema()) 657 .bind(draft.scope().map(|scope| scope.as_bytes().to_vec())) 658 .execute(&mut **transaction) 659 .await 660 .map_err(map_backend)?; 661 Ok(()) 662 } 663 664 async fn append_tx( 665 transaction: &mut sqlx::Transaction<'_, sqlx::Sqlite>, 666 draft: &AuthoredDraft, 667 expected_head: Option<AuthoredDraftRevision>, 668 ) -> Result<DraftAppendDisposition, Error> { 669 if let Some(row) = bounded_row::load( 670 &mut **transaction, 671 draft.draft_id().as_bytes(), 672 Some(draft.revision()), 673 ) 674 .await? 675 { 676 let existing = decode_row(&row)?; 677 return if existing == *draft { 678 Ok(DraftAppendDisposition::Replay) 679 } else { 680 Err(Error::DraftRevisionConflict) 681 }; 682 } 683 684 let head = bounded_row::load(&mut **transaction, draft.draft_id().as_bytes(), None) 685 .await? 686 .as_ref() 687 .map(decode_row) 688 .transpose()?; 689 match (head.as_ref(), expected_head) { 690 (None, None) if draft.revision() == AuthoredDraftRevision::INITIAL => {} 691 (Some(previous), Some(expected)) if previous.revision() == expected => { 692 draft.validate_successor_of(previous)?; 693 } 694 _ => { 695 return Err(Error::DraftRevisionConflict); 696 } 697 } 698 699 insert_draft_tx(transaction, draft).await?; 700 Ok(DraftAppendDisposition::Inserted) 701 } 702 703 #[cfg(test)] 704 #[path = "authored_draft_pair_tests.rs"] 705 mod pair_tests;