keyring_worker.rs (30165B)
1 use std::sync::atomic::{AtomicU8, Ordering}; 2 use std::sync::{Arc, Mutex}; 3 use std::thread::JoinHandle; 4 use std::time::Duration; 5 6 use harvestcircle_application::{BoxFuture, DurableRequestId, SecretStore}; 7 use harvestcircle_domain::{PublicKey, SafeError, SafeErrorCode, SafeMessage, SecretKeyInput}; 8 use tokio::sync::{oneshot, watch}; 9 10 const KEYRING_QUEUE_CAPACITY: usize = 8; 11 const KEYRING_SHUTDOWN_DEADLINE: Duration = Duration::from_secs(30); 12 const OPERATION_QUEUED: u8 = 0; 13 const OPERATION_STARTED: u8 = 1; 14 const OPERATION_COMPLETED: u8 = 2; 15 const OPERATION_CANCELLED: u8 = 3; 16 17 enum Request { 18 Put( 19 DurableRequestId, 20 PublicKey, 21 SecretKeyInput, 22 Arc<AtomicU8>, 23 oneshot::Sender<Result<(), SafeError>>, 24 ), 25 Verify( 26 DurableRequestId, 27 PublicKey, 28 SecretKeyInput, 29 Arc<AtomicU8>, 30 oneshot::Sender<Result<(), SafeError>>, 31 ), 32 Load( 33 PublicKey, 34 oneshot::Sender<Result<SecretKeyInput, SafeError>>, 35 ), 36 Contains( 37 PublicKey, 38 Arc<AtomicU8>, 39 oneshot::Sender<Result<bool, SafeError>>, 40 ), 41 Delete( 42 DurableRequestId, 43 PublicKey, 44 Arc<AtomicU8>, 45 oneshot::Sender<Result<(), SafeError>>, 46 ), 47 Close, 48 } 49 50 struct CancellationGuard { 51 phase: Arc<AtomicU8>, 52 armed: bool, 53 } 54 55 impl CancellationGuard { 56 fn new(phase: Arc<AtomicU8>) -> Self { 57 Self { phase, armed: true } 58 } 59 60 fn disarm(&mut self) { 61 self.armed = false; 62 } 63 } 64 65 impl Drop for CancellationGuard { 66 fn drop(&mut self) { 67 if self.armed { 68 let _ = self.phase.compare_exchange( 69 OPERATION_QUEUED, 70 OPERATION_CANCELLED, 71 Ordering::AcqRel, 72 Ordering::Acquire, 73 ); 74 } 75 } 76 } 77 78 pub(crate) struct BoundedKeyringWorker { 79 sender: Mutex<Option<std::sync::mpsc::SyncSender<Request>>>, 80 completion: watch::Receiver<bool>, 81 thread: Mutex<Option<JoinHandle<()>>>, 82 } 83 84 impl BoundedKeyringWorker { 85 pub(crate) fn new(store: impl SecretStore + 'static) -> Result<Arc<Self>, SafeError> { 86 let (sender, receiver) = std::sync::mpsc::sync_channel(KEYRING_QUEUE_CAPACITY); 87 let (completion_sender, completion_receiver) = watch::channel(false); 88 let runtime = tokio::runtime::Builder::new_current_thread() 89 .enable_all() 90 .build() 91 .map_err(|_| worker_unavailable())?; 92 let thread = std::thread::Builder::new() 93 .name("harvestcircle-keyring-worker".to_owned()) 94 .spawn(move || { 95 while let Ok(request) = receiver.recv() { 96 match request { 97 Request::Put(request_id, public_key, secret, phase, response) => { 98 if start_operation(&phase) { 99 let result = runtime.block_on(async { 100 store.put(&request_id, public_key, secret).await 101 }); 102 finish_operation(&phase); 103 let _ = response.send(result); 104 } 105 } 106 Request::Verify(request_id, public_key, secret, phase, response) => { 107 if start_operation(&phase) { 108 let result = runtime.block_on(async { 109 store.verify(&request_id, public_key, secret).await 110 }); 111 finish_operation(&phase); 112 let _ = response.send(result); 113 } 114 } 115 Request::Load(public_key, response) => { 116 let _ = response 117 .send(runtime.block_on(async { store.load(public_key).await })); 118 } 119 Request::Contains(public_key, phase, response) => { 120 if start_operation(&phase) { 121 let result = 122 runtime.block_on(async { store.contains(public_key).await }); 123 finish_operation(&phase); 124 let _ = response.send(result); 125 } 126 } 127 Request::Delete(request_id, public_key, phase, response) => { 128 if start_operation(&phase) { 129 let result = runtime.block_on(async { 130 store.delete(&request_id, public_key).await 131 }); 132 finish_operation(&phase); 133 let _ = response.send(result); 134 } 135 } 136 Request::Close => break, 137 } 138 } 139 let _ = completion_sender.send(true); 140 }) 141 .map_err(|_| worker_unavailable())?; 142 Ok(Arc::new(Self { 143 sender: Mutex::new(Some(sender)), 144 completion: completion_receiver, 145 thread: Mutex::new(Some(thread)), 146 })) 147 } 148 149 async fn submit<T>( 150 &self, 151 request: impl FnOnce(Arc<AtomicU8>, oneshot::Sender<T>) -> Request, 152 ) -> Result<T, SafeError> { 153 let (response_sender, response_receiver) = oneshot::channel(); 154 let phase = Arc::new(AtomicU8::new(OPERATION_QUEUED)); 155 let mut cancellation = CancellationGuard::new(Arc::clone(&phase)); 156 { 157 let sender_guard = self.sender.lock().map_err(|_| worker_unavailable())?; 158 let Some(sender) = sender_guard.as_ref() else { 159 return Err(worker_unavailable()); 160 }; 161 sender 162 .try_send(request(Arc::clone(&phase), response_sender)) 163 .map_err(|_| worker_unavailable())?; 164 } 165 let response = response_receiver.await; 166 cancellation.disarm(); 167 response.map_err(|_| match phase.load(Ordering::Acquire) { 168 OPERATION_STARTED | OPERATION_COMPLETED => recovery_required(), 169 _ => worker_unavailable(), 170 }) 171 } 172 173 pub(crate) async fn close(&self) -> Result<(), SafeError> { 174 self.close_with_deadline(KEYRING_SHUTDOWN_DEADLINE).await 175 } 176 177 async fn close_with_deadline(&self, deadline: Duration) -> Result<(), SafeError> { 178 let sender = self.sender.lock().map_err(|_| worker_unavailable())?.take(); 179 if let Some(sender) = sender { 180 signal_close(sender); 181 } 182 let mut completion = self.completion.clone(); 183 tokio::time::timeout(deadline, async { 184 while !*completion.borrow() { 185 completion 186 .changed() 187 .await 188 .map_err(|_| worker_unavailable())?; 189 } 190 loop { 191 let finished = self 192 .thread 193 .lock() 194 .map_err(|_| worker_unavailable())? 195 .as_ref() 196 .is_none_or(JoinHandle::is_finished); 197 if finished { 198 break; 199 } 200 tokio::task::yield_now().await; 201 } 202 Ok::<(), SafeError>(()) 203 }) 204 .await 205 .map_err(|_| recovery_required())??; 206 let thread = self.thread.lock().map_err(|_| worker_unavailable())?.take(); 207 if let Some(thread) = thread { 208 thread.join().map_err(|_| recovery_required())?; 209 } 210 Ok(()) 211 } 212 } 213 214 fn start_operation(phase: &AtomicU8) -> bool { 215 phase 216 .compare_exchange( 217 OPERATION_QUEUED, 218 OPERATION_STARTED, 219 Ordering::AcqRel, 220 Ordering::Acquire, 221 ) 222 .is_ok() 223 } 224 225 fn finish_operation(phase: &AtomicU8) { 226 phase.store(OPERATION_COMPLETED, Ordering::Release); 227 } 228 229 fn signal_close(sender: std::sync::mpsc::SyncSender<Request>) { 230 match sender.try_send(Request::Close) { 231 Ok(()) 232 | Err(std::sync::mpsc::TrySendError::Full(Request::Close)) 233 | Err(std::sync::mpsc::TrySendError::Disconnected(Request::Close)) => {} 234 Err( 235 std::sync::mpsc::TrySendError::Full(_) | std::sync::mpsc::TrySendError::Disconnected(_), 236 ) => { 237 unreachable!("close signaling constructs only close requests") 238 } 239 } 240 } 241 242 impl SecretStore for BoundedKeyringWorker { 243 fn put<'a>( 244 &'a self, 245 request_id: &'a DurableRequestId, 246 public_key: PublicKey, 247 secret: SecretKeyInput, 248 ) -> BoxFuture<'a, Result<(), SafeError>> { 249 Box::pin(async move { 250 self.submit(|phase, response| { 251 Request::Put(request_id.clone(), public_key, secret, phase, response) 252 }) 253 .await? 254 }) 255 } 256 257 fn verify<'a>( 258 &'a self, 259 request_id: &'a DurableRequestId, 260 public_key: PublicKey, 261 secret: SecretKeyInput, 262 ) -> BoxFuture<'a, Result<(), SafeError>> { 263 Box::pin(async move { 264 self.submit(|phase, response| { 265 Request::Verify(request_id.clone(), public_key, secret, phase, response) 266 }) 267 .await? 268 }) 269 } 270 271 fn load(&self, public_key: PublicKey) -> BoxFuture<'_, Result<SecretKeyInput, SafeError>> { 272 Box::pin(async move { 273 self.submit(|_phase, response| Request::Load(public_key, response)) 274 .await? 275 }) 276 } 277 278 fn contains(&self, public_key: PublicKey) -> BoxFuture<'_, Result<bool, SafeError>> { 279 Box::pin(async move { 280 self.submit(|phase, response| Request::Contains(public_key, phase, response)) 281 .await? 282 }) 283 } 284 285 fn delete<'a>( 286 &'a self, 287 request_id: &'a DurableRequestId, 288 public_key: PublicKey, 289 ) -> BoxFuture<'a, Result<(), SafeError>> { 290 Box::pin(async move { 291 self.submit(|phase, response| { 292 Request::Delete(request_id.clone(), public_key, phase, response) 293 }) 294 .await? 295 }) 296 } 297 } 298 299 impl Drop for BoundedKeyringWorker { 300 fn drop(&mut self) { 301 if let Ok(sender) = self.sender.get_mut() 302 && let Some(sender) = sender.take() 303 { 304 let _ = sender.try_send(Request::Close); 305 } 306 } 307 } 308 309 const fn worker_unavailable() -> SafeError { 310 SafeError::new( 311 SafeErrorCode::KeyringUnavailable, 312 SafeMessage::new("The operating system credential store is unavailable."), 313 ) 314 } 315 316 const fn recovery_required() -> SafeError { 317 SafeError::new( 318 SafeErrorCode::PendingOperationRecoveryRequired, 319 SafeMessage::new("Credential operation recovery is required."), 320 ) 321 } 322 323 #[cfg(test)] 324 mod tests { 325 use std::sync::Arc; 326 use std::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering}; 327 use std::time::Duration; 328 329 use harvestcircle_application::{ 330 BoxFuture, DurableRequestId, InMemorySecretStore, SecretStore, 331 }; 332 use harvestcircle_domain::{PublicKey, SafeError, SafeErrorCode, SecretKeyInput}; 333 use tokio::sync::oneshot; 334 335 use super::{ 336 BoundedKeyringWorker, CancellationGuard, KEYRING_QUEUE_CAPACITY, OPERATION_CANCELLED, 337 OPERATION_COMPLETED, OPERATION_QUEUED, OPERATION_STARTED, Request, finish_operation, 338 signal_close, start_operation, 339 }; 340 341 fn public_key() -> PublicKey { 342 PublicKey::from_hex("7e7e9c42a91bfef19fa7ea99d52d8afdb67d893a8fefba1f5cb9793f2107f6d7") 343 .expect("public key") 344 } 345 346 fn request_id() -> DurableRequestId { 347 DurableRequestId::parse("01890f3e-7b1c-7000-8000-000000000249").expect("request") 348 } 349 350 fn alternate_request_id() -> DurableRequestId { 351 DurableRequestId::parse("01890f3e-7b1c-7000-8000-000000000250").expect("request") 352 } 353 354 fn secret() -> SecretKeyInput { 355 SecretKeyInput::parse( 356 "0000000000000000000000000000000000000000000000000000000000000001".to_owned(), 357 ) 358 .expect("secret") 359 } 360 361 struct BlockingPutState { 362 inner: InMemorySecretStore, 363 block_next_put: AtomicBool, 364 put_started: AtomicBool, 365 release_put: AtomicBool, 366 put_calls: AtomicUsize, 367 verify_calls: AtomicUsize, 368 block_next_verify: AtomicBool, 369 verify_started: AtomicBool, 370 release_verify: AtomicBool, 371 } 372 373 #[derive(Clone)] 374 struct BlockingPutStore { 375 state: Arc<BlockingPutState>, 376 } 377 378 impl BlockingPutStore { 379 fn new() -> Self { 380 Self { 381 state: Arc::new(BlockingPutState { 382 inner: InMemorySecretStore::default(), 383 block_next_put: AtomicBool::new(true), 384 put_started: AtomicBool::new(false), 385 release_put: AtomicBool::new(false), 386 put_calls: AtomicUsize::new(0), 387 verify_calls: AtomicUsize::new(0), 388 block_next_verify: AtomicBool::new(false), 389 verify_started: AtomicBool::new(false), 390 release_verify: AtomicBool::new(false), 391 }), 392 } 393 } 394 395 async fn wait_until_started(&self) { 396 while !self.state.put_started.load(Ordering::Acquire) { 397 tokio::task::yield_now().await; 398 } 399 } 400 401 fn release(&self) { 402 self.state.release_put.store(true, Ordering::Release); 403 } 404 405 fn put_calls(&self) -> usize { 406 self.state.put_calls.load(Ordering::Acquire) 407 } 408 409 async fn wait_until_verification_started(&self) { 410 while !self.state.verify_started.load(Ordering::Acquire) { 411 tokio::task::yield_now().await; 412 } 413 } 414 415 fn release_verification(&self) { 416 self.state.release_verify.store(true, Ordering::Release); 417 } 418 419 async fn contains_direct(&self, public_key: PublicKey) -> bool { 420 self.state 421 .inner 422 .contains(public_key) 423 .await 424 .expect("contains") 425 } 426 } 427 428 impl SecretStore for BlockingPutStore { 429 fn put<'a>( 430 &'a self, 431 request_id: &'a DurableRequestId, 432 public_key: PublicKey, 433 secret: SecretKeyInput, 434 ) -> BoxFuture<'a, Result<(), SafeError>> { 435 Box::pin(async move { 436 self.state.put_calls.fetch_add(1, Ordering::AcqRel); 437 if self.state.block_next_put.swap(false, Ordering::AcqRel) { 438 self.state.put_started.store(true, Ordering::Release); 439 while !self.state.release_put.load(Ordering::Acquire) { 440 std::thread::yield_now(); 441 } 442 } 443 self.state.inner.put(request_id, public_key, secret).await 444 }) 445 } 446 447 fn verify<'a>( 448 &'a self, 449 request_id: &'a DurableRequestId, 450 public_key: PublicKey, 451 secret: SecretKeyInput, 452 ) -> BoxFuture<'a, Result<(), SafeError>> { 453 Box::pin(async move { 454 self.state.verify_calls.fetch_add(1, Ordering::AcqRel); 455 if self.state.block_next_verify.swap(false, Ordering::AcqRel) { 456 self.state.verify_started.store(true, Ordering::Release); 457 while !self.state.release_verify.load(Ordering::Acquire) { 458 std::thread::yield_now(); 459 } 460 } 461 self.state 462 .inner 463 .verify(request_id, public_key, secret) 464 .await 465 }) 466 } 467 468 fn load(&self, public_key: PublicKey) -> BoxFuture<'_, Result<SecretKeyInput, SafeError>> { 469 self.state.inner.load(public_key) 470 } 471 472 fn contains(&self, public_key: PublicKey) -> BoxFuture<'_, Result<bool, SafeError>> { 473 self.state.inner.contains(public_key) 474 } 475 476 fn delete<'a>( 477 &'a self, 478 request_id: &'a DurableRequestId, 479 public_key: PublicKey, 480 ) -> BoxFuture<'a, Result<(), SafeError>> { 481 self.state.inner.delete(request_id, public_key) 482 } 483 } 484 485 struct SlowContainsStore { 486 inner: InMemorySecretStore, 487 } 488 489 impl SecretStore for SlowContainsStore { 490 fn put<'a>( 491 &'a self, 492 request_id: &'a DurableRequestId, 493 public_key: PublicKey, 494 secret: SecretKeyInput, 495 ) -> BoxFuture<'a, Result<(), SafeError>> { 496 self.inner.put(request_id, public_key, secret) 497 } 498 499 fn load(&self, public_key: PublicKey) -> BoxFuture<'_, Result<SecretKeyInput, SafeError>> { 500 self.inner.load(public_key) 501 } 502 503 fn contains(&self, public_key: PublicKey) -> BoxFuture<'_, Result<bool, SafeError>> { 504 Box::pin(async move { 505 std::thread::sleep(Duration::from_millis(50)); 506 self.inner.contains(public_key).await 507 }) 508 } 509 510 fn delete<'a>( 511 &'a self, 512 request_id: &'a DurableRequestId, 513 public_key: PublicKey, 514 ) -> BoxFuture<'a, Result<(), SafeError>> { 515 self.inner.delete(request_id, public_key) 516 } 517 } 518 519 #[tokio::test] 520 async fn worker_round_trips_without_exposing_secret_material() { 521 let worker = BoundedKeyringWorker::new(InMemorySecretStore::default()).expect("worker"); 522 worker 523 .put(&request_id(), public_key(), secret()) 524 .await 525 .expect("put"); 526 assert!(worker.contains(public_key()).await.expect("contains")); 527 let loaded = worker.load(public_key()).await.expect("load"); 528 assert_eq!(loaded.with_exposed_secret(str::len), 64); 529 worker 530 .delete(&request_id(), public_key()) 531 .await 532 .expect("delete"); 533 worker.close().await.expect("close"); 534 assert!(worker.contains(public_key()).await.is_err()); 535 } 536 537 #[tokio::test] 538 async fn worker_readonly_verification_forwards_full_binding_and_closes_without_mutation() { 539 let store = BlockingPutStore::new(); 540 store.release(); 541 let worker = BoundedKeyringWorker::new(store.clone()).expect("worker"); 542 worker 543 .put(&request_id(), public_key(), secret()) 544 .await 545 .expect("put"); 546 let exact = worker.verify(&request_id(), public_key(), secret()).await; 547 let changed_request = worker 548 .verify(&alternate_request_id(), public_key(), secret()) 549 .await 550 .expect_err("request mismatch"); 551 let changed_secret = worker 552 .verify( 553 &request_id(), 554 public_key(), 555 SecretKeyInput::parse( 556 "0000000000000000000000000000000000000000000000000000000000000002".to_owned(), 557 ) 558 .expect("different secret"), 559 ) 560 .await 561 .expect_err("secret mismatch"); 562 let missing = worker 563 .verify( 564 &request_id(), 565 PublicKey::from_bytes([8; 32]).expect("other public key"), 566 secret(), 567 ) 568 .await 569 .expect_err("missing credential"); 570 let retained = worker 571 .load(public_key()) 572 .await 573 .expect("retained credential"); 574 worker.close().await.expect("close after verification"); 575 let closed = worker 576 .verify(&request_id(), public_key(), secret()) 577 .await 578 .expect_err("closed worker"); 579 assert!(exact.is_ok()); 580 assert_eq!( 581 changed_request.code(), 582 SafeErrorCode::InvalidApplicationState 583 ); 584 assert_eq!( 585 changed_secret.code(), 586 SafeErrorCode::InvalidApplicationState 587 ); 588 assert_eq!(missing.code(), SafeErrorCode::CredentialMissing); 589 assert_eq!(closed.code(), SafeErrorCode::KeyringUnavailable); 590 assert_eq!(store.put_calls(), 1); 591 assert_eq!(store.state.verify_calls.load(Ordering::Acquire), 4); 592 assert!(retained.with_exposed_secret(|value| { 593 secret().with_exposed_secret(|expected| value == expected) 594 })); 595 assert!(worker.thread.lock().expect("thread").is_none()); 596 assert!(*worker.completion.borrow()); 597 let public_evidence = 598 format!("{changed_request:?} {changed_secret:?} {missing:?} {closed:?}"); 599 assert!(!secret().with_exposed_secret(|value| public_evidence.contains(value))); 600 } 601 602 #[tokio::test(flavor = "multi_thread", worker_threads = 2)] 603 async fn queued_readonly_verification_cancellation_skips_adapter_and_joins_on_close() { 604 let store = BlockingPutStore::new(); 605 let worker = BoundedKeyringWorker::new(store.clone()).expect("worker"); 606 let first_worker = Arc::clone(&worker); 607 let first = tokio::spawn(async move { 608 first_worker 609 .put(&request_id(), public_key(), secret()) 610 .await 611 }); 612 store.wait_until_started().await; 613 let request = request_id(); 614 let mut verification = worker.verify(&request, public_key(), secret()); 615 tokio::select! { 616 biased; 617 result = &mut verification => panic!("blocked worker completed verification: {result:?}"), 618 () = tokio::task::yield_now() => {} 619 } 620 drop(verification); 621 store.release(); 622 let first_result = first.await.expect("first task"); 623 worker 624 .close() 625 .await 626 .expect("close drains cancelled verification"); 627 assert!(first_result.is_ok()); 628 assert_eq!(store.put_calls(), 1); 629 assert_eq!(store.state.verify_calls.load(Ordering::Acquire), 0); 630 assert!(store.contains_direct(public_key()).await); 631 assert!(worker.thread.lock().expect("thread").is_none()); 632 assert!(*worker.completion.borrow()); 633 } 634 635 #[tokio::test(flavor = "multi_thread", worker_threads = 2)] 636 async fn started_readonly_verification_caller_loss_keeps_close_resumable_until_join() { 637 let store = BlockingPutStore::new(); 638 store.release(); 639 let worker = BoundedKeyringWorker::new(store.clone()).expect("worker"); 640 worker 641 .put(&request_id(), public_key(), secret()) 642 .await 643 .expect("put"); 644 store.state.block_next_verify.store(true, Ordering::Release); 645 let verification_worker = Arc::clone(&worker); 646 let verification = tokio::spawn(async move { 647 verification_worker 648 .verify(&request_id(), public_key(), secret()) 649 .await 650 }); 651 store.wait_until_verification_started().await; 652 verification.abort(); 653 let caller_loss = verification 654 .await 655 .expect_err("cancelled verification caller"); 656 let timeout = worker.close_with_deadline(Duration::from_millis(1)).await; 657 let retained_thread = worker.thread.lock().expect("thread").is_some(); 658 store.release_verification(); 659 let resumed_close = worker.close_with_deadline(Duration::from_secs(1)).await; 660 let retained = store 661 .state 662 .inner 663 .load(public_key()) 664 .await 665 .expect("retained credential"); 666 assert!(caller_loss.is_cancelled()); 667 assert_eq!( 668 timeout 669 .expect_err("started verification keeps close pending") 670 .code(), 671 SafeErrorCode::PendingOperationRecoveryRequired 672 ); 673 assert!(retained_thread); 674 assert!(resumed_close.is_ok()); 675 assert_eq!(store.put_calls(), 1); 676 assert_eq!(store.state.verify_calls.load(Ordering::Acquire), 1); 677 assert!(retained.with_exposed_secret(|value| { 678 secret().with_exposed_secret(|expected| value == expected) 679 })); 680 assert!(worker.thread.lock().expect("thread").is_none()); 681 assert!(*worker.completion.borrow()); 682 } 683 684 #[test] 685 fn operation_phases_are_closed_and_cancel_only_queued_work() { 686 let cancelled = Arc::new(AtomicU8::new(OPERATION_QUEUED)); 687 drop(CancellationGuard::new(Arc::clone(&cancelled))); 688 assert_eq!(cancelled.load(Ordering::Acquire), OPERATION_CANCELLED); 689 assert!(!start_operation(&cancelled)); 690 691 let completed = AtomicU8::new(OPERATION_QUEUED); 692 assert!(start_operation(&completed)); 693 assert_eq!(completed.load(Ordering::Acquire), OPERATION_STARTED); 694 finish_operation(&completed); 695 assert_eq!(completed.load(Ordering::Acquire), OPERATION_COMPLETED); 696 697 let started = Arc::new(AtomicU8::new(OPERATION_STARTED)); 698 drop(CancellationGuard::new(Arc::clone(&started))); 699 assert_eq!(started.load(Ordering::Acquire), OPERATION_STARTED); 700 } 701 702 #[tokio::test(flavor = "multi_thread", worker_threads = 2)] 703 async fn cancellation_before_start_has_no_credential_effect() { 704 let store = BlockingPutStore::new(); 705 let worker = BoundedKeyringWorker::new(store.clone()).expect("worker"); 706 let first_worker = Arc::clone(&worker); 707 let first = tokio::spawn(async move { 708 first_worker 709 .put(&request_id(), public_key(), secret()) 710 .await 711 }); 712 store.wait_until_started().await; 713 714 let queued_worker = Arc::clone(&worker); 715 let queued = tokio::spawn(async move { 716 queued_worker 717 .put(&alternate_request_id(), public_key(), secret()) 718 .await 719 }); 720 tokio::task::yield_now().await; 721 queued.abort(); 722 assert!(queued.await.expect_err("cancelled task").is_cancelled()); 723 724 store.release(); 725 first.await.expect("first task").expect("first put"); 726 worker.close().await.expect("close"); 727 assert_eq!(store.put_calls(), 1); 728 } 729 730 #[tokio::test(flavor = "multi_thread", worker_threads = 2)] 731 async fn caller_loss_after_start_preserves_unknown_outcome_for_recovery() { 732 let store = BlockingPutStore::new(); 733 let worker = BoundedKeyringWorker::new(store.clone()).expect("worker"); 734 let operation_worker = Arc::clone(&worker); 735 let operation = tokio::spawn(async move { 736 operation_worker 737 .put(&request_id(), public_key(), secret()) 738 .await 739 }); 740 store.wait_until_started().await; 741 operation.abort(); 742 assert!(operation.await.expect_err("cancelled task").is_cancelled()); 743 744 store.release(); 745 while !store.contains_direct(public_key()).await { 746 tokio::task::yield_now().await; 747 } 748 worker.close().await.expect("close"); 749 assert_eq!(store.put_calls(), 1); 750 } 751 752 #[tokio::test(flavor = "multi_thread", worker_threads = 2)] 753 async fn shutdown_timeout_is_recovery_required_and_retry_joins_thread() { 754 let store = BlockingPutStore::new(); 755 let worker = BoundedKeyringWorker::new(store.clone()).expect("worker"); 756 let operation_worker = Arc::clone(&worker); 757 let operation = tokio::spawn(async move { 758 operation_worker 759 .put(&request_id(), public_key(), secret()) 760 .await 761 }); 762 store.wait_until_started().await; 763 764 let timeout = worker 765 .close_with_deadline(Duration::from_millis(1)) 766 .await 767 .expect_err("blocked worker must time out"); 768 assert_eq!( 769 timeout.code(), 770 SafeErrorCode::PendingOperationRecoveryRequired 771 ); 772 assert!(worker.thread.lock().expect("thread").is_some()); 773 774 store.release(); 775 operation.await.expect("operation task").expect("put"); 776 worker 777 .close_with_deadline(Duration::from_secs(1)) 778 .await 779 .expect("retry close"); 780 assert!(worker.thread.lock().expect("thread").is_none()); 781 assert!(*worker.completion.borrow()); 782 } 783 784 #[tokio::test(flavor = "current_thread")] 785 async fn response_waiting_never_blocks_the_tokio_runtime_thread() { 786 let worker = BoundedKeyringWorker::new(SlowContainsStore { 787 inner: InMemorySecretStore::default(), 788 }) 789 .expect("worker"); 790 let response = worker.contains(public_key()); 791 tokio::pin!(response); 792 793 tokio::select! { 794 biased; 795 result = &mut response => panic!("slow keyring response completed before runtime progress: {result:?}"), 796 () = tokio::task::yield_now() => {} 797 } 798 799 assert!(!response.await.expect("contains")); 800 worker.close().await.expect("close"); 801 } 802 803 #[test] 804 fn close_signal_never_blocks_on_a_full_bounded_queue() { 805 let (sender, receiver) = std::sync::mpsc::sync_channel(KEYRING_QUEUE_CAPACITY); 806 for _ in 0..KEYRING_QUEUE_CAPACITY { 807 let (response, _response_receiver) = oneshot::channel(); 808 let phase = Arc::new(AtomicU8::new(OPERATION_QUEUED)); 809 assert!( 810 sender 811 .try_send(Request::Contains(public_key(), phase, response)) 812 .is_ok() 813 ); 814 } 815 let (overflow_response, _overflow_receiver) = oneshot::channel(); 816 assert!(matches!( 817 sender.try_send(Request::Contains( 818 public_key(), 819 Arc::new(AtomicU8::new(OPERATION_QUEUED)), 820 overflow_response, 821 )), 822 Err(std::sync::mpsc::TrySendError::Full(_)) 823 )); 824 825 signal_close(sender); 826 827 for _ in 0..KEYRING_QUEUE_CAPACITY { 828 assert!(matches!(receiver.recv(), Ok(Request::Contains(_, _, _)))); 829 } 830 assert!(receiver.recv().is_err()); 831 } 832 }