handle most of the remaining unwraps · seoul.systems/microcosm-rs@42c61ff

+41 -67

1 changed file

expand all

link_aggregator

src

storage

rocks_store.rs

+41 -67

link_aggregator/src/storage/rocks_store.rs

··· 452 452 self.prefix_iter_cf(&cf, TargetIdTargetPrefix(target.clone())) 453 453 } 454 454 455 - fn check_for_did_dups(&self, problem_did_id: &DidId) { 456 - let cf = self.db.cf_handle(DID_IDS_CF).unwrap(); 457 - let mut seen_ids: HashMap<DidId, Vec<Did>> = HashMap::new(); 458 - for (k, v) in self 459 - .db 460 - .iterator_cf(&cf, rocksdb::IteratorMode::Start) 461 - .map_while(Result::ok) 462 - { 463 - let did: Did = _kr(&k).unwrap(); 464 - let DidIdValue(did_id, _) = _vr(&v).unwrap(); 465 - if let Some(dids) = seen_ids.get_mut(&did_id) { 466 - let is_problem = did_id == *problem_did_id; 467 - eprintln!( 468 - "dup! problem? {is_problem}. at did_id {did_id:?}: {dids:?} and now {did:?}" 469 - ); 470 - dids.push(did); 471 - } else { 472 - if did_id == *problem_did_id { 473 - eprintln!("found our friend {did:?} {did_id:?}"); 474 - } 475 - assert!(seen_ids.insert(did_id, vec![did]).is_none()); 476 - } 477 - } 478 - eprintln!("done checking."); 479 - } 480 - 481 455 // 482 456 // higher-level event action handlers 483 457 // 484 458 485 - fn add_links(&mut self, record_id: &RecordId, links: &[CollectedLink], batch: &mut WriteBatch) { 486 - let DidIdValue(did_id, _) = self 487 - .did_id_table 488 - .get_or_create_id_val(&self.db, batch, &record_id.did) 489 - .unwrap(); 459 + fn add_links( 460 + &mut self, 461 + record_id: &RecordId, 462 + links: &[CollectedLink], 463 + batch: &mut WriteBatch, 464 + ) -> Result<()> { 465 + let DidIdValue(did_id, _) = 466 + self.did_id_table 467 + .get_or_create_id_val(&self.db, batch, &record_id.did)?; 490 468 491 469 let record_link_key = RecordLinkKey( 492 470 did_id, ··· 501 479 Collection(record_id.collection()), 502 480 RPath(path.clone()), 503 481 ); 504 - let target_id = self 505 - .target_id_table 506 - .get_or_create_id_val(&self.db, batch, &target_key) 507 - .unwrap(); 482 + let target_id = 483 + self.target_id_table 484 + .get_or_create_id_val(&self.db, batch, &target_key)?; 508 485 self.merge_target_linker(batch, &target_id, &did_id, &RKey(record_id.rkey())); 509 486 510 487 record_link_targets.add(RecordLinkTarget(RPath(path.clone()), target_id)) 511 488 } 512 489 513 490 self.put_link_targets(batch, &record_link_key, &record_link_targets); 491 + Ok(()) 514 492 } 515 493 516 - fn remove_links(&mut self, record_id: &RecordId, batch: &mut WriteBatch) { 517 - let Some(DidIdValue(linking_did_id, did_active)) = self 518 - .did_id_table 519 - .get_id_val(&self.db, &record_id.did) 520 - .unwrap() 494 + fn remove_links(&mut self, record_id: &RecordId, batch: &mut WriteBatch) -> Result<()> { 495 + let Some(DidIdValue(linking_did_id, did_active)) = 496 + self.did_id_table.get_id_val(&self.db, &record_id.did)? 521 497 else { 522 - return; // we don't know her: nothing to do 498 + return Ok(()); // we don't know her: nothing to do 523 499 }; 524 500 if !did_active { 525 501 eprintln!( ··· 533 509 Collection(record_id.collection()), 534 510 RKey(record_id.rkey()), 535 511 ); 536 - let Some(record_link_targets) = self.get_record_link_targets(&record_link_key).unwrap() 537 - else { 538 - return; // we don't have these links 512 + let Some(record_link_targets) = self.get_record_link_targets(&record_link_key)? else { 513 + return Ok(()); // we don't have these links 539 514 }; 540 515 541 516 // we do read -> modify -> write here: could merge-op in the deletes instead? ··· 558 533 panic!("reverse index didn't have us"); 559 534 } 560 535 Some(linkers) 561 - }).unwrap(); 536 + })?; 562 537 } 563 538 564 539 self.delete_record_link(batch, &record_link_key); 540 + Ok(()) 565 541 } 566 542 567 543 fn update_links( ··· 569 545 record_id: &RecordId, 570 546 new_links: &[CollectedLink], 571 547 batch: &mut WriteBatch, 572 - ) { 573 - self.remove_links(record_id, batch); 574 - self.add_links(record_id, new_links, batch); 548 + ) -> Result<()> { 549 + self.remove_links(record_id, batch)?; 550 + self.add_links(record_id, new_links, batch)?; 551 + Ok(()) 575 552 } 576 553 577 - fn set_account(&mut self, did: &Did, active: bool, batch: &mut WriteBatch) { 554 + fn set_account(&mut self, did: &Did, active: bool, batch: &mut WriteBatch) -> Result<()> { 578 555 // this needs to be read-modify-write since the did_id needs to stay the same, 579 556 // which has a benefit of allowing to avoid adding entries for dids we don't 580 557 // need. reading on dids needs to be cheap anyway for the current design, and 581 558 // did active/inactive updates are low-freq in the firehose so, eh, it's fine. 582 559 self.update_did_id_value(batch, did, |current_value| { 583 560 Some(DidIdValue(current_value.did_id(), active)) 584 - }) 585 - .unwrap(); 561 + })?; 562 + Ok(()) 586 563 } 587 564 588 - fn delete_account(&mut self, did: &Did, batch: &mut WriteBatch) { 589 - let Some(DidIdValue(did_id, active)) = self.did_id_table.get_id_val(&self.db, did).unwrap() 590 - else { 591 - return; // ignore updates for dids we don't know about 565 + fn delete_account(&mut self, did: &Did, batch: &mut WriteBatch) -> Result<()> { 566 + let Some(DidIdValue(did_id, active)) = self.did_id_table.get_id_val(&self.db, did)? else { 567 + return Ok(()); // ignore updates for dids we don't know about 592 568 }; 593 569 self.delete_did_id_value(batch, did); 594 570 // TODO: also delete the reverse!! ··· 622 598 eprintln!("from record link key {record_link_key:?}"); 623 599 eprintln!("but could not find this link :/"); 624 600 eprintln!("checking for did_id dups..."); 625 - self.check_for_did_dups(&did_id); 626 601 eprintln!("ok so what the heck. did_id again, for did {did:?}:"); 627 602 let did_id_again = self 628 603 .did_id_table 629 604 .get_id_val(&self.db, did) 630 605 .unwrap() 631 - .unwrap(); 606 + .unwrap(); // TODO 632 607 eprintln!("did_id_value (again): {did_id_again:?}"); 633 608 panic!("ohnoooo"); 634 609 } 635 610 Some(linkers) 636 - }) 637 - .unwrap(); 611 + })?; 638 612 } 639 613 } 640 - 641 - self.db.write(mini_batch).unwrap(); // todo 614 + self.db.write(mini_batch)?; // todo 642 615 } 616 + Ok(()) 643 617 } 644 618 } 645 619 ··· 677 651 let mut batch = WriteBatch::default(); 678 652 match event { 679 653 ActionableEvent::CreateLinks { record_id, links } => { 680 - self.add_links(record_id, links, &mut batch) 654 + self.add_links(record_id, links, &mut batch)? 681 655 } 682 656 ActionableEvent::UpdateLinks { 683 657 record_id, 684 658 new_links, 685 - } => self.update_links(record_id, new_links, &mut batch), 686 - ActionableEvent::DeleteRecord(record_id) => self.remove_links(record_id, &mut batch), 687 - ActionableEvent::ActivateAccount(did) => self.set_account(did, true, &mut batch), 688 - ActionableEvent::DeactivateAccount(did) => self.set_account(did, false, &mut batch), 689 - ActionableEvent::DeleteAccount(did) => self.delete_account(did, &mut batch), 659 + } => self.update_links(record_id, new_links, &mut batch)?, 660 + ActionableEvent::DeleteRecord(record_id) => self.remove_links(record_id, &mut batch)?, 661 + ActionableEvent::ActivateAccount(did) => self.set_account(did, true, &mut batch)?, 662 + ActionableEvent::DeactivateAccount(did) => self.set_account(did, false, &mut batch)?, 663 + ActionableEvent::DeleteAccount(did) => self.delete_account(did, &mut batch)?, 690 664 } 691 665 batch.put(JETSTREAM_CURSOR_KEY.as_bytes(), _rv(cursor)); 692 666 self.db.write(batch)?;

Configure Feed

Configure Feed