db: decode sequenced event rows leniently · goat.navy/tranquil-pds@8d7f891

+122 -83

1 changed file

Expand all

crates

tranquil-db

src

postgres

repo.rs

+122 -83

crates/tranquil-db/src/postgres/repo.rs

View file

··· 10 10 use tranquil_types::{AtUri, CidLink, Did, Handle, Nsid, Rkey, Tid}; 11 11 use uuid::Uuid; 12 12 13 + use super::col; 13 14 use super::user::map_sqlx_error; 15 + use super::{column, column_vec, legacy_column, opt_column}; 14 16 15 17 struct RecordRow { 16 18 rkey: String, ··· 43 45 ) -> Result<Option<EventBlocks>, DbError> { 44 46 match (block_cids, block_data) { 45 47 (Some(cids), Some(data)) if cids.len() == data.len() => match cids.is_empty() { 46 - true => Ok(legacy_fallback(legacy_blocks_cids)), 48 + true => legacy_fallback(legacy_blocks_cids), 47 49 false => Ok(Some(EventBlocks::Inline( 48 50 cids.into_iter() 49 51 .zip(data) ··· 57 59 (Some(_), None) | (None, Some(_)) => Err(DbError::CorruptData( 58 60 "repo_seq.block_cids/block_data partially populated", 59 61 )), 60 - (None, None) => Ok(legacy_fallback(legacy_blocks_cids)), 62 + (None, None) => legacy_fallback(legacy_blocks_cids), 61 63 } 62 64 } 63 65 64 - fn legacy_fallback(legacy_blocks_cids: Option<Vec<String>>) -> Option<EventBlocks> { 66 + fn legacy_fallback( 67 + legacy_blocks_cids: Option<Vec<String>>, 68 + ) -> Result<Option<EventBlocks>, DbError> { 65 69 match legacy_blocks_cids { 66 - Some(cids) if !cids.is_empty() => Some(EventBlocks::LegacyCids( 67 - cids.into_iter().map(CidLink::from).collect(), 68 - )), 69 - _ => None, 70 + Some(cids) if !cids.is_empty() => Ok(Some(EventBlocks::LegacyCids(column_vec( 71 + cids, 72 + col::REPO_SEQ_BLOCKS_CIDS, 73 + )?))), 74 + _ => Ok(None), 70 75 } 71 76 } 72 77 ··· 97 102 let blocks = row_to_event_blocks(r.block_cids, r.block_data, r.blocks_cids)?; 98 103 Ok(SequencedEvent { 99 104 seq: r.seq.into(), 100 - did: Did::from(r.did), 105 + did: column(r.did, col::REPO_SEQ_DID)?, 101 106 created_at: r.created_at, 102 107 event_type: r.event_type, 103 - commit_cid: r.commit_cid.map(CidLink::from), 104 - prev_cid: r.prev_cid.map(CidLink::from), 105 - prev_data_cid: r.prev_data_cid.map(CidLink::from), 108 + commit_cid: opt_column(r.commit_cid, col::REPO_SEQ_COMMIT_CID)?, 109 + prev_cid: opt_column(r.prev_cid, col::REPO_SEQ_PREV_CID)?, 110 + prev_data_cid: opt_column(r.prev_data_cid, col::REPO_SEQ_PREV_DATA_CID)?, 106 111 ops: r.ops, 107 112 blobs: r 108 113 .blobs 109 - .map(|blobs| blobs.into_iter().map(CidLink::from).collect()), 114 + .map(|blobs| column_vec(blobs, col::REPO_SEQ_BLOBS)) 115 + .transpose()?, 110 116 blocks, 111 - handle: r.handle.map(Handle::from), 117 + handle: r 118 + .handle 119 + .and_then(|h| legacy_column(h, col::REPO_SEQ_HANDLE)), 112 120 active: r.active, 113 121 status, 114 - rev: r.rev.map(Tid::from), 122 + rev: opt_column(r.rev, col::REPO_SEQ_REV)?, 115 123 }) 124 + } 125 + 126 + fn collect_sequenced_rows(rows: Vec<SequencedEventRow>) -> Vec<SequencedEvent> { 127 + rows.into_iter() 128 + .filter_map(|r| { 129 + let seq = r.seq; 130 + map_sequenced_row(r) 131 + .inspect_err(|e| { 132 + tracing::error!(seq, error = %e, "skipping a repo_seq row that doesn't decode"); 133 + }) 134 + .ok() 135 + }) 136 + .collect() 116 137 } 117 138 118 139 const SEQUENCER_LOCK_KEY: i64 = 0x0074_7261_6e73_6571; ··· 266 287 .await 267 288 .map_err(map_sqlx_error)?; 268 289 269 - Ok(result.map(CidLink::from)) 290 + opt_column(result, col::REPOS_REPO_ROOT_CID) 270 291 } 271 292 272 293 async fn get_repo(&self, user_id: Uuid) -> Result<Option<RepoInfo>, DbError> { ··· 278 299 .await 279 300 .map_err(map_sqlx_error)?; 280 301 281 - Ok(row.map(|r| RepoInfo { 282 - user_id: r.user_id, 283 - repo_root_cid: CidLink::from(r.repo_root_cid), 284 - repo_rev: r.repo_rev.map(Tid::from), 285 - })) 302 + row.map(|r| { 303 + Ok(RepoInfo { 304 + user_id: r.user_id, 305 + repo_root_cid: column(r.repo_root_cid, col::REPOS_REPO_ROOT_CID)?, 306 + repo_rev: opt_column(r.repo_rev, col::REPOS_REPO_REV)?, 307 + }) 308 + }) 309 + .transpose() 286 310 } 287 311 288 312 async fn get_repo_root_by_did(&self, did: &Did) -> Result<Option<CidLink>, DbError> { ··· 294 318 .await 295 319 .map_err(map_sqlx_error)?; 296 320 297 - Ok(result.map(CidLink::from)) 321 + opt_column(result, col::REPOS_REPO_ROOT_CID) 298 322 } 299 323 300 324 async fn count_repos(&self) -> Result<i64, DbError> { ··· 312 336 .await 313 337 .map_err(map_sqlx_error)?; 314 338 315 - Ok(rows 316 - .into_iter() 317 - .map(|r| RepoWithoutRev { 318 - user_id: r.user_id, 319 - repo_root_cid: CidLink::from(r.repo_root_cid), 339 + rows.into_iter() 340 + .map(|r| { 341 + Ok(RepoWithoutRev { 342 + user_id: r.user_id, 343 + repo_root_cid: column(r.repo_root_cid, col::REPOS_REPO_ROOT_CID)?, 344 + }) 320 345 }) 321 - .collect()) 346 + .collect() 322 347 } 323 348 324 349 async fn upsert_records( ··· 405 430 .await 406 431 .map_err(map_sqlx_error)?; 407 432 408 - Ok(result.map(CidLink::from)) 433 + opt_column(result, col::RECORDS_RECORD_CID) 409 434 } 410 435 411 436 async fn list_records( ··· 418 443 rkey_start: Option<&Rkey>, 419 444 rkey_end: Option<&Rkey>, 420 445 ) -> Result<Vec<RecordInfo>, DbError> { 421 - let to_record_info = |rows: Vec<RecordRow>| { 422 - rows.into_iter() 423 - .map(|r| RecordInfo { 424 - rkey: Rkey::from(r.rkey), 425 - record_cid: CidLink::from(r.record_cid), 446 + let to_record_info = |rows: Vec<RecordRow>| -> Result<Vec<RecordInfo>, DbError> { 447 + Ok(rows 448 + .into_iter() 449 + .filter_map(|r| { 450 + Some(RecordInfo { 451 + rkey: legacy_column(r.rkey, col::RECORDS_RKEY)?, 452 + record_cid: legacy_column(r.record_cid, col::RECORDS_RECORD_CID)?, 453 + }) 426 454 }) 427 - .collect() 455 + .collect()) 428 456 }; 429 457 430 458 let collection_str = collection.as_str(); ··· 446 474 .fetch_all(&self.pool) 447 475 .await 448 476 .map_err(map_sqlx_error)?; 449 - Ok(to_record_info(rows)) 477 + to_record_info(rows) 450 478 } 451 479 true => { 452 480 let rows = sqlx::query_as!( ··· 462 490 .fetch_all(&self.pool) 463 491 .await 464 492 .map_err(map_sqlx_error)?; 465 - Ok(to_record_info(rows)) 493 + to_record_info(rows) 466 494 } 467 495 }; 468 496 } ··· 486 514 .fetch_all(&self.pool) 487 515 .await 488 516 .map_err(map_sqlx_error)?; 489 - Ok(to_record_info(rows)) 517 + to_record_info(rows) 490 518 } 491 519 true => { 492 520 let rows = sqlx::query_as!( ··· 503 531 .fetch_all(&self.pool) 504 532 .await 505 533 .map_err(map_sqlx_error)?; 506 - Ok(to_record_info(rows)) 534 + to_record_info(rows) 507 535 } 508 536 }; 509 537 } ··· 525 553 .fetch_all(&self.pool) 526 554 .await 527 555 .map_err(map_sqlx_error)?; 528 - Ok(to_record_info(rows)) 556 + to_record_info(rows) 529 557 } 530 558 true => { 531 559 let rows = sqlx::query_as!( ··· 541 569 .fetch_all(&self.pool) 542 570 .await 543 571 .map_err(map_sqlx_error)?; 544 - Ok(to_record_info(rows)) 572 + to_record_info(rows) 545 573 } 546 574 }; 547 575 } ··· 563 591 .fetch_all(&self.pool) 564 592 .await 565 593 .map_err(map_sqlx_error)?; 566 - Ok(to_record_info(rows)) 594 + to_record_info(rows) 567 595 } 568 596 true => { 569 597 let rows = sqlx::query_as!( ··· 579 607 .fetch_all(&self.pool) 580 608 .await 581 609 .map_err(map_sqlx_error)?; 582 - Ok(to_record_info(rows)) 610 + to_record_info(rows) 583 611 } 584 612 }; 585 613 } ··· 598 626 .fetch_all(&self.pool) 599 627 .await 600 628 .map_err(map_sqlx_error)?; 601 - Ok(to_record_info(rows)) 629 + to_record_info(rows) 602 630 } 603 631 true => { 604 632 let rows = sqlx::query_as!( ··· 613 641 .fetch_all(&self.pool) 614 642 .await 615 643 .map_err(map_sqlx_error)?; 616 - Ok(to_record_info(rows)) 644 + to_record_info(rows) 617 645 } 618 646 } 619 647 } ··· 629 657 630 658 Ok(rows 631 659 .into_iter() 632 - .map(|r| FullRecordInfo { 633 - collection: Nsid::from(r.collection), 634 - rkey: Rkey::from(r.rkey), 635 - record_cid: CidLink::from(r.record_cid), 660 + .filter_map(|r| { 661 + Some(FullRecordInfo { 662 + collection: legacy_column(r.collection, col::RECORDS_COLLECTION)?, 663 + rkey: legacy_column(r.rkey, col::RECORDS_RKEY)?, 664 + record_cid: legacy_column(r.record_cid, col::RECORDS_RECORD_CID)?, 665 + }) 636 666 }) 637 667 .collect()) 638 668 } ··· 646 676 .await 647 677 .map_err(map_sqlx_error)?; 648 678 649 - Ok(rows.into_iter().map(Nsid::from).collect()) 679 + Ok(rows 680 + .into_iter() 681 + .filter_map(|c| legacy_column(c, col::RECORDS_COLLECTION)) 682 + .collect()) 650 683 } 651 684 652 685 async fn count_records(&self, repo_id: Uuid) -> Result<i64, DbError> { ··· 1024 1057 .await 1025 1058 .map_err(map_sqlx_error)?; 1026 1059 1027 - Ok(row.map(|r| RepoAccountInfo { 1028 - user_id: r.id, 1029 - did: Did::from(r.did), 1030 - deactivated_at: r.deactivated_at, 1031 - takedown_ref: r.takedown_ref, 1032 - repo_root_cid: r.repo_root_cid.map(CidLink::from), 1033 - })) 1060 + row.map(|r| { 1061 + Ok(RepoAccountInfo { 1062 + user_id: r.id, 1063 + did: column(r.did, col::USERS_DID)?, 1064 + deactivated_at: r.deactivated_at, 1065 + takedown_ref: r.takedown_ref, 1066 + repo_root_cid: opt_column(r.repo_root_cid, col::REPOS_REPO_ROOT_CID)?, 1067 + }) 1068 + }) 1069 + .transpose() 1034 1070 } 1035 1071 1036 1072 async fn get_events_since_seq( ··· 1054 1090 .fetch_all(&self.pool) 1055 1091 .await 1056 1092 .map_err(map_sqlx_error)?; 1057 - rows.into_iter().map(map_sequenced_row).collect() 1093 + Ok(collect_sequenced_rows(rows)) 1058 1094 } 1059 1095 None => { 1060 1096 let rows = sqlx::query_as!( ··· 1069 1105 .fetch_all(&self.pool) 1070 1106 .await 1071 1107 .map_err(map_sqlx_error)?; 1072 - rows.into_iter().map(map_sequenced_row).collect() 1108 + Ok(collect_sequenced_rows(rows)) 1073 1109 } 1074 1110 } 1075 1111 } ··· 1092 1128 .fetch_all(&self.pool) 1093 1129 .await 1094 1130 .map_err(map_sqlx_error)?; 1095 - rows.into_iter().map(map_sequenced_row).collect() 1131 + Ok(collect_sequenced_rows(rows)) 1096 1132 } 1097 1133 1098 1134 async fn get_event_by_seq( ··· 1132 1168 .fetch_all(&self.pool) 1133 1169 .await 1134 1170 .map_err(map_sqlx_error)?; 1135 - rows.into_iter().map(map_sequenced_row).collect() 1171 + Ok(collect_sequenced_rows(rows)) 1136 1172 } 1137 1173 1138 1174 async fn list_repos_paginated( ··· 1155 1191 .await 1156 1192 .map_err(map_sqlx_error)?; 1157 1193 1158 - Ok(rows 1159 - .into_iter() 1160 - .map(|r| RepoListItem { 1161 - did: Did::from(r.did), 1162 - deactivated_at: r.deactivated_at, 1163 - takedown_ref: r.takedown_ref, 1164 - repo_root_cid: CidLink::from(r.repo_root_cid), 1165 - repo_rev: r.repo_rev.map(Tid::from), 1194 + rows.into_iter() 1195 + .map(|r| { 1196 + Ok(RepoListItem { 1197 + did: column(r.did, col::USERS_DID)?, 1198 + deactivated_at: r.deactivated_at, 1199 + takedown_ref: r.takedown_ref, 1200 + repo_root_cid: column(r.repo_root_cid, col::REPOS_REPO_ROOT_CID)?, 1201 + repo_rev: opt_column(r.repo_rev, col::REPOS_REPO_REV)?, 1202 + }) 1166 1203 }) 1167 - .collect()) 1204 + .collect() 1168 1205 } 1169 1206 1170 1207 async fn get_repo_root_cid_by_user_id( ··· 1178 1215 .fetch_optional(&self.pool) 1179 1216 .await 1180 1217 .map_err(map_sqlx_error)?; 1181 - Ok(cid.map(CidLink::from)) 1218 + opt_column(cid, col::REPOS_REPO_ROOT_CID) 1182 1219 } 1183 1220 1184 1221 async fn import_repo_data( ··· 1520 1557 .await 1521 1558 .map_err(map_sqlx_error)?; 1522 1559 1523 - Ok(rows 1524 - .into_iter() 1525 - .map(|(user_id, repo_root_cid, repo_rev)| UserWithoutBlocks { 1526 - user_id, 1527 - repo_root_cid: CidLink::from(repo_root_cid), 1528 - repo_rev: repo_rev.map(Tid::from), 1560 + rows.into_iter() 1561 + .map(|(user_id, repo_root_cid, repo_rev)| { 1562 + Ok(UserWithoutBlocks { 1563 + user_id, 1564 + repo_root_cid: column(repo_root_cid, col::REPOS_REPO_ROOT_CID)?, 1565 + repo_rev: opt_column(repo_rev, col::REPOS_REPO_REV)?, 1566 + }) 1529 1567 }) 1530 - .collect()) 1568 + .collect() 1531 1569 } 1532 1570 1533 1571 async fn get_users_needing_record_blobs_backfill( ··· 1548 1586 .await 1549 1587 .map_err(map_sqlx_error)?; 1550 1588 1551 - Ok(rows 1552 - .into_iter() 1553 - .map(|r| UserNeedingRecordBlobsBackfill { 1554 - user_id: r.user_id, 1555 - did: Did::from(r.did), 1589 + rows.into_iter() 1590 + .map(|r| { 1591 + Ok(UserNeedingRecordBlobsBackfill { 1592 + user_id: r.user_id, 1593 + did: column(r.did, col::USERS_DID)?, 1594 + }) 1556 1595 }) 1557 - .collect()) 1596 + .collect() 1558 1597 } 1559 1598 1560 1599 async fn insert_record_blobs(

Configure Feed

Configure Feed