use std::collections::HashSet;use std::ops::Index;use std::sync::Arc; use anyhow::anyhow;use chrono::{DateTime, Duration, Utc};use cloud_object_client::MockObjectClient;use cloud_objects::cloud_object::ServerPermissions;use firebase::FirebaseError;use itertools::Itertools;use warpui::{App, Entity, ModelHandle, SingletonEntity}; use super::QueueDependency;use crate::ai::facts::{AIFact, AIMemory, CloudAIFactModel};use crate::cloud_object::model::actions::{ ObjectAction, ObjectActionHistory, ObjectActionSubtype, ObjectActionType,};use crate::cloud_object::model::generic_string_model::GenericStringObjectId;use crate::cloud_object::{ CloudModelType, CloudObjectEventEntrypoint, CreateCloudObjectResult, CreatedCloudObject, GenericStringObjectFormat, JsonObjectType, ObjectIdType, ObjectType, Owner, Revision, RevisionAndLastEditor, ServerCreationInfo, UpdateCloudObjectResult,};use crate::drive::CloudObjectTypeAndId;use crate::notebooks::{CloudNotebookModel, NotebookId};use crate::server::cloud_objects::update_manager::InitiatedBy;use crate::server::ids::{ClientId, HashableId, ServerId, ServerIdAndType, SyncId};use crate::server::server_api::ServerApiProvider;use crate::server::server_api::auth::UserAuthenticationError;use crate::server::sync_queue::{ CreationFailureReason, GenericStringObjectToCreate, QueueItemId, SerializedModel, SyncQueueEvent,};use crate::system::SystemStats;use crate::test_util::assert_eventually;use crate::workflows::CloudWorkflowModel;use crate::workflows::workflow::{Argument, ArgumentType, Workflow};use crate::{NetworkStatus, QueueItem, SyncQueue}; #[derive(Default)]struct Events(Vec<SyncQueueEvent>); impl Entity for Events { type Event = ();} impl Index<usize> for Events { type Output = SyncQueueEvent; fn index(&self, index: usize) -> &Self::Output { &self.0[index] }} fn initialize_app(app: &mut App) { app.add_singleton_model(|_| ServerApiProvider::new_for_test()); app.add_singleton_model(|_| NetworkStatus::new()); app.add_singleton_model(|_| SystemStats::new());} fn queue_item_dependencies<const N: usize>( queue_item_ids: [QueueItemId; N],) -> HashSet<QueueDependency> { queue_item_ids .into_iter() .map(QueueDependency::QueueItem) .collect()} fn bulk_create_dependency(queue_item_id: QueueItemId, client_id: ClientId) -> QueueDependency { QueueDependency::BulkCreateGenericStringObject { queue_item_id, client_id, }} fn create_sync_queue( app: &mut App, queue_items: Vec<QueueItem>, cloud_objects_client_mock: MockObjectClient, immediately_start_dequeueing: bool,) -> ModelHandle<SyncQueue> { let sync_queue = app.add_singleton_model(|ctx| { SyncQueue::new(queue_items, Arc::new(cloud_objects_client_mock), ctx) }); if immediately_start_dequeueing { SyncQueue::handle(app).update(app, |sync_queue, ctx| { sync_queue.start_dequeueing(ctx); }); } sync_queue} #[test]fn test_create_notebook() { App::test((), |mut app| async move { let notebook_id = ClientId::default(); let notebook_server_id = ServerId::from(1); let notebook_ts = DateTime::<Utc>::default(); let notebook = CloudNotebookModel { title: "Shared Notebook".to_string(), data: "Hello".to_string(), ai_document_id: None, conversation_id: None, }; let mut cloud_objects_client_mock = MockObjectClient::new(); cloud_objects_client_mock .expect_create_notebook() .times(1) .return_once(move |_| { Ok(CreateCloudObjectResult::Success { created_cloud_object: CreatedCloudObject { client_id: notebook_id, revision_and_editor: RevisionAndLastEditor { revision: notebook_ts.into(), last_editor_uid: None, }, metadata_ts: notebook_ts.into(), server_id_and_type: ServerIdAndType { id: notebook_server_id, id_type: ObjectIdType::Notebook, }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, }) }); initialize_app(&mut app); let sync_queue = create_sync_queue(&mut app, vec![], cloud_objects_client_mock, true); let sync_queue_events = app.add_model(|_ctx| Events::default()); sync_queue_events.update(&mut app, |_, ctx| { ctx.subscribe_to_model(&sync_queue, |me, _, event, _ctx| me.0.push(event.clone())) }); // Enqueue a single item and wait for the response to complete. sync_queue .update(&mut app, |sync_queue, ctx| { sync_queue.enqueue( QueueItem::CreateObject { object_type: ObjectType::Notebook, owner: Owner::mock_current_user(), id: notebook_id, title: None, serialized_model: Some(Arc::new(notebook.serialized())), initial_folder_id: None, entrypoint: Default::default(), initiated_by: InitiatedBy::User, }, ctx, ); ctx.await_spawned_future(sync_queue.spawned_futures[0]) }) .await; sync_queue_events.update(&mut app, |sync_queue_events, _ctx| { assert_eq!(sync_queue_events.0.len(), 1); assert_eq!( sync_queue_events[0], SyncQueueEvent::ObjectCreationSuccessful { server_creation_info: ServerCreationInfo { server_id_and_type: ServerIdAndType { id: notebook_server_id, id_type: ObjectIdType::Notebook }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, client_id: notebook_id, revision_and_editor: RevisionAndLastEditor { revision: notebook_ts.into(), last_editor_uid: None }, metadata_ts: notebook_ts.into(), initiated_by: InitiatedBy::User } ); }) });} #[test]fn test_generic_string_object_unique_key_failure() { App::test((), |mut app| async move { let owner = Owner::mock_current_user(); let gso_id = ClientId::default(); let gso_json = "{\"storage_key\":\"somepref\",\"value\":true,\"platform\":\"Global\"}"; let workflow_id = ClientId::default(); let workflow_server_id = ServerId::from(1); let workflow_ts = DateTime::<Utc>::default(); let workflow_data = Workflow::new("my workflow", "echo hi"); // This pattern is used in a couple places to control the order of async operations. // Basically, we wait to receive a message on a channel before certain mocks can continue, // so that we have time to assert intermediate states, at which point we fire a message. let (tx, rx) = std::sync::mpsc::channel(); initialize_app(&mut app); let mut cloud_objects_client_mock = MockObjectClient::new(); cloud_objects_client_mock .expect_create_generic_string_object() .times(1) .return_once(move |_, _, _| { Ok(CreateCloudObjectResult::GenericStringObjectUniqueKeyConflict) }); cloud_objects_client_mock .expect_create_workflow() .times(1) .return_once(move |_| { rx.recv().unwrap(); Ok(CreateCloudObjectResult::Success { created_cloud_object: CreatedCloudObject { client_id: workflow_id, revision_and_editor: RevisionAndLastEditor { revision: workflow_ts.into(), last_editor_uid: None, }, metadata_ts: workflow_ts.into(), server_id_and_type: ServerIdAndType { id: workflow_server_id, id_type: ObjectIdType::Workflow, }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, }) }); let sync_queue = create_sync_queue( &mut app, vec![ QueueItem::CreateObject { object_type: ObjectType::GenericStringObject(GenericStringObjectFormat::Json( JsonObjectType::Preference, )), owner, id: gso_id, title: None, serialized_model: Some(Arc::new(gso_json.to_string().into())), initial_folder_id: None, entrypoint: CloudObjectEventEntrypoint::Unknown, initiated_by: InitiatedBy::User, }, QueueItem::CreateWorkflow { object_type: ObjectType::Workflow, owner, id: workflow_id, model: Arc::new(CloudWorkflowModel { data: workflow_data, }), initial_folder_id: None, entrypoint: CloudObjectEventEntrypoint::Unknown, initiated_by: InitiatedBy::User, }, ], cloud_objects_client_mock, false, ); let sync_queue_events = app.add_model(|_ctx| Events::default()); sync_queue_events.update(&mut app, |_, ctx| { ctx.subscribe_to_model(&sync_queue, |me, _, event, _ctx| me.0.push(event.clone())) }); // Await the first future. sync_queue .update(&mut app, |queue, ctx| { queue.start_dequeueing(ctx); ctx.await_spawned_future(queue.spawned_futures[0]) }) .await; app.read(|ctx| { // The GSO failure should not stop the queue. assert!(sync_queue.as_ref(ctx).is_dequeueing()); let events = sync_queue_events.as_ref(ctx); assert_eq!(events.0.len(), 1); assert_eq!( &events.0, &[SyncQueueEvent::ObjectCreationFailure { reason: CreationFailureReason::UniqueKeyConflict { id: gso_id.to_hash(), initiated_by: InitiatedBy::User }, },] ); }); // Allow the workflow to be created. tx.send(()).unwrap(); sync_queue .update(&mut app, |queue, ctx| { assert!(queue.is_dequeueing()); ctx.await_spawned_future(queue.spawned_futures[1]) }) .await; app.read(|ctx| { // The queue should still be dequeueing. assert!(sync_queue.as_ref(ctx).is_dequeueing()); let events = sync_queue_events.as_ref(ctx); assert_eq!(events.0.len(), 2); assert_eq!( &events.0, &[ SyncQueueEvent::ObjectCreationFailure { reason: CreationFailureReason::UniqueKeyConflict { id: gso_id.to_hash(), initiated_by: InitiatedBy::User }, }, SyncQueueEvent::ObjectCreationSuccessful { server_creation_info: ServerCreationInfo { server_id_and_type: ServerIdAndType { id: workflow_server_id, id_type: ObjectIdType::Workflow }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, client_id: workflow_id, revision_and_editor: RevisionAndLastEditor { revision: workflow_ts.into(), last_editor_uid: None }, metadata_ts: workflow_ts.into(), initiated_by: InitiatedBy::User } ] ); }); });} #[test]fn test_dequeue_after_transient_failure() { App::test((), |mut app| async move { let owner = Owner::mock_current_user(); let success_notebook_id = ClientId::default(); let success_notebook_server_id = ServerId::from(1); let success_notebook_ts = DateTime::<Utc>::default(); let success_notebook = CloudNotebookModel { title: "Successful Notebook".to_string(), data: "Hello :)".to_string(), ai_document_id: None, conversation_id: None, }; let failure_notebook_id = ClientId::default(); let failure_notebook = CloudNotebookModel { title: "Failed Notebook".to_string(), data: "Hello :(".to_string(), ai_document_id: None, conversation_id: None, }; let (tx, rx) = std::sync::mpsc::channel(); let failure_attempts = 1 + super::DEFAULT_RETRY_OPTION.remaining_retries(); let mut cloud_objects_client_mock = MockObjectClient::new(); cloud_objects_client_mock .expect_create_notebook() .times(failure_attempts) .returning(move |_| Err(anyhow!("Transient network error!!!!"))); cloud_objects_client_mock .expect_create_notebook() .times(1) .return_once(move |_| { rx.recv().unwrap(); Ok(CreateCloudObjectResult::Success { created_cloud_object: CreatedCloudObject { client_id: success_notebook_id, revision_and_editor: RevisionAndLastEditor { revision: success_notebook_ts.into(), last_editor_uid: None, }, metadata_ts: success_notebook_ts.into(), server_id_and_type: ServerIdAndType { id: success_notebook_server_id, id_type: ObjectIdType::Notebook, }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, }) }); initialize_app(&mut app); let sync_queue = create_sync_queue( &mut app, vec![ QueueItem::CreateObject { object_type: ObjectType::Notebook, owner, id: failure_notebook_id, title: None, serialized_model: Some(Arc::new(failure_notebook.serialized())), initial_folder_id: None, entrypoint: CloudObjectEventEntrypoint::Unknown, initiated_by: InitiatedBy::User, }, QueueItem::CreateObject { object_type: ObjectType::Notebook, owner, id: success_notebook_id, title: None, serialized_model: Some(Arc::new(success_notebook.serialized())), initial_folder_id: None, entrypoint: CloudObjectEventEntrypoint::Unknown, initiated_by: InitiatedBy::User, }, ], cloud_objects_client_mock, false, ); let sync_queue_events = app.add_model(|_ctx| Events::default()); sync_queue_events.update(&mut app, |_, ctx| { ctx.subscribe_to_model(&sync_queue, |me, _, event, _ctx| me.0.push(event.clone())) }); sync_queue .update(&mut app, |queue, ctx| { queue.start_dequeueing(ctx); ctx.await_spawned_future(queue.spawned_futures[0]) }) .await; // Wait for the first notebook's creation to fail. // The failures are retried, but their futures are spawned on background threads, // so we can't access them. Instead, we wait for a SyncQueue event to appear in the model. assert_eventually!( 4000 => sync_queue_events.read(&app, |events, _ctx| !events.0.is_empty()), "Timed out waiting for failure" ); sync_queue_events.read(&app, |events, _| { assert_eq!( &events.0[0], &SyncQueueEvent::ObjectCreationFailure { reason: CreationFailureReason::Other { id: failure_notebook_id.to_hash(), initiated_by: InitiatedBy::User }, } ); }); // Allow the second notebook to be created. tx.send(()).unwrap(); sync_queue .update(&mut app, |queue, ctx| { assert!(queue.is_dequeueing()); ctx.await_spawned_future(queue.spawned_futures[1]) }) .await; sync_queue_events.read(&app, |events, _| { assert_eq!(events.0.len(), 2); assert_eq!( &events.0, &[ SyncQueueEvent::ObjectCreationFailure { reason: CreationFailureReason::Other { id: failure_notebook_id.to_hash(), initiated_by: InitiatedBy::User }, }, SyncQueueEvent::ObjectCreationSuccessful { server_creation_info: ServerCreationInfo { server_id_and_type: ServerIdAndType { id: success_notebook_server_id, id_type: ObjectIdType::Notebook }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, client_id: success_notebook_id, revision_and_editor: RevisionAndLastEditor { revision: success_notebook_ts.into(), last_editor_uid: None }, metadata_ts: success_notebook_ts.into(), initiated_by: InitiatedBy::User } ] ); }); });} Lines 1–500 of 1826. Load more
use std::collections::HashSet; use std::ops::Index; use std::sync::Arc; use anyhow::anyhow; use chrono::{DateTime, Duration, Utc}; use cloud_object_client::MockObjectClient; use cloud_objects::cloud_object::ServerPermissions; use firebase::FirebaseError; use itertools::Itertools; use warpui::{App, Entity, ModelHandle, SingletonEntity}; use super::QueueDependency; use crate::ai::facts::{AIFact, AIMemory, CloudAIFactModel}; use crate::cloud_object::model::actions::{ ObjectAction, ObjectActionHistory, ObjectActionSubtype, ObjectActionType, }; use crate::cloud_object::model::generic_string_model::GenericStringObjectId; use crate::cloud_object::{ CloudModelType, CloudObjectEventEntrypoint, CreateCloudObjectResult, CreatedCloudObject, GenericStringObjectFormat, JsonObjectType, ObjectIdType, ObjectType, Owner, Revision, RevisionAndLastEditor, ServerCreationInfo, UpdateCloudObjectResult, }; use crate::drive::CloudObjectTypeAndId; use crate::notebooks::{CloudNotebookModel, NotebookId}; use crate::server::cloud_objects::update_manager::InitiatedBy; use crate::server::ids::{ClientId, HashableId, ServerId, ServerIdAndType, SyncId}; use crate::server::server_api::ServerApiProvider; use crate::server::server_api::auth::UserAuthenticationError; use crate::server::sync_queue::{ CreationFailureReason, GenericStringObjectToCreate, QueueItemId, SerializedModel, SyncQueueEvent, }; use crate::system::SystemStats; use crate::test_util::assert_eventually; use crate::workflows::CloudWorkflowModel; use crate::workflows::workflow::{Argument, ArgumentType, Workflow}; use crate::{NetworkStatus, QueueItem, SyncQueue}; #[derive(Default)] struct Events(Vec<SyncQueueEvent>); impl Entity for Events { type Event = (); } impl Index<usize> for Events { type Output = SyncQueueEvent; fn index(&self, index: usize) -> &Self::Output { &self.0[index] } } fn initialize_app(app: &mut App) { app.add_singleton_model(|_| ServerApiProvider::new_for_test()); app.add_singleton_model(|_| NetworkStatus::new()); app.add_singleton_model(|_| SystemStats::new()); } fn queue_item_dependencies<const N: usize>( queue_item_ids: [QueueItemId; N], ) -> HashSet<QueueDependency> { queue_item_ids .into_iter() .map(QueueDependency::QueueItem) .collect() } fn bulk_create_dependency(queue_item_id: QueueItemId, client_id: ClientId) -> QueueDependency { QueueDependency::BulkCreateGenericStringObject { queue_item_id, client_id, } } fn create_sync_queue( app: &mut App, queue_items: Vec<QueueItem>, cloud_objects_client_mock: MockObjectClient, immediately_start_dequeueing: bool, ) -> ModelHandle<SyncQueue> { let sync_queue = app.add_singleton_model(|ctx| { SyncQueue::new(queue_items, Arc::new(cloud_objects_client_mock), ctx) }); if immediately_start_dequeueing { SyncQueue::handle(app).update(app, |sync_queue, ctx| { sync_queue.start_dequeueing(ctx); }); } sync_queue } #[test] fn test_create_notebook() { App::test((), |mut app| async move { let notebook_id = ClientId::default(); let notebook_server_id = ServerId::from(1); let notebook_ts = DateTime::<Utc>::default(); let notebook = CloudNotebookModel { title: "Shared Notebook".to_string(), data: "Hello".to_string(), ai_document_id: None, conversation_id: None, }; let mut cloud_objects_client_mock = MockObjectClient::new(); cloud_objects_client_mock .expect_create_notebook() .times(1) .return_once(move |_| { Ok(CreateCloudObjectResult::Success { created_cloud_object: CreatedCloudObject { client_id: notebook_id, revision_and_editor: RevisionAndLastEditor { revision: notebook_ts.into(), last_editor_uid: None, }, metadata_ts: notebook_ts.into(), server_id_and_type: ServerIdAndType { id: notebook_server_id, id_type: ObjectIdType::Notebook, }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, }) }); initialize_app(&mut app); let sync_queue = create_sync_queue(&mut app, vec![], cloud_objects_client_mock, true); let sync_queue_events = app.add_model(|_ctx| Events::default()); sync_queue_events.update(&mut app, |_, ctx| { ctx.subscribe_to_model(&sync_queue, |me, _, event, _ctx| me.0.push(event.clone())) }); // Enqueue a single item and wait for the response to complete. sync_queue .update(&mut app, |sync_queue, ctx| { sync_queue.enqueue( QueueItem::CreateObject { object_type: ObjectType::Notebook, owner: Owner::mock_current_user(), id: notebook_id, title: None, serialized_model: Some(Arc::new(notebook.serialized())), initial_folder_id: None, entrypoint: Default::default(), initiated_by: InitiatedBy::User, }, ctx, ); ctx.await_spawned_future(sync_queue.spawned_futures[0]) }) .await; sync_queue_events.update(&mut app, |sync_queue_events, _ctx| { assert_eq!(sync_queue_events.0.len(), 1); assert_eq!( sync_queue_events[0], SyncQueueEvent::ObjectCreationSuccessful { server_creation_info: ServerCreationInfo { server_id_and_type: ServerIdAndType { id: notebook_server_id, id_type: ObjectIdType::Notebook }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, client_id: notebook_id, revision_and_editor: RevisionAndLastEditor { revision: notebook_ts.into(), last_editor_uid: None }, metadata_ts: notebook_ts.into(), initiated_by: InitiatedBy::User } ); }) }); } #[test] fn test_generic_string_object_unique_key_failure() { App::test((), |mut app| async move { let owner = Owner::mock_current_user(); let gso_id = ClientId::default(); let gso_json = "{\"storage_key\":\"somepref\",\"value\":true,\"platform\":\"Global\"}"; let workflow_id = ClientId::default(); let workflow_server_id = ServerId::from(1); let workflow_ts = DateTime::<Utc>::default(); let workflow_data = Workflow::new("my workflow", "echo hi"); // This pattern is used in a couple places to control the order of async operations. // Basically, we wait to receive a message on a channel before certain mocks can continue, // so that we have time to assert intermediate states, at which point we fire a message. let (tx, rx) = std::sync::mpsc::channel(); initialize_app(&mut app); let mut cloud_objects_client_mock = MockObjectClient::new(); cloud_objects_client_mock .expect_create_generic_string_object() .times(1) .return_once(move |_, _, _| { Ok(CreateCloudObjectResult::GenericStringObjectUniqueKeyConflict) }); cloud_objects_client_mock .expect_create_workflow() .times(1) .return_once(move |_| { rx.recv().unwrap(); Ok(CreateCloudObjectResult::Success { created_cloud_object: CreatedCloudObject { client_id: workflow_id, revision_and_editor: RevisionAndLastEditor { revision: workflow_ts.into(), last_editor_uid: None, }, metadata_ts: workflow_ts.into(), server_id_and_type: ServerIdAndType { id: workflow_server_id, id_type: ObjectIdType::Workflow, }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, }) }); let sync_queue = create_sync_queue( &mut app, vec![ QueueItem::CreateObject { object_type: ObjectType::GenericStringObject(GenericStringObjectFormat::Json( JsonObjectType::Preference, )), owner, id: gso_id, title: None, serialized_model: Some(Arc::new(gso_json.to_string().into())), initial_folder_id: None, entrypoint: CloudObjectEventEntrypoint::Unknown, initiated_by: InitiatedBy::User, }, QueueItem::CreateWorkflow { object_type: ObjectType::Workflow, owner, id: workflow_id, model: Arc::new(CloudWorkflowModel { data: workflow_data, }), initial_folder_id: None, entrypoint: CloudObjectEventEntrypoint::Unknown, initiated_by: InitiatedBy::User, }, ], cloud_objects_client_mock, false, ); let sync_queue_events = app.add_model(|_ctx| Events::default()); sync_queue_events.update(&mut app, |_, ctx| { ctx.subscribe_to_model(&sync_queue, |me, _, event, _ctx| me.0.push(event.clone())) }); // Await the first future. sync_queue .update(&mut app, |queue, ctx| { queue.start_dequeueing(ctx); ctx.await_spawned_future(queue.spawned_futures[0]) }) .await; app.read(|ctx| { // The GSO failure should not stop the queue. assert!(sync_queue.as_ref(ctx).is_dequeueing()); let events = sync_queue_events.as_ref(ctx); assert_eq!(events.0.len(), 1); assert_eq!( &events.0, &[SyncQueueEvent::ObjectCreationFailure { reason: CreationFailureReason::UniqueKeyConflict { id: gso_id.to_hash(), initiated_by: InitiatedBy::User }, },] ); }); // Allow the workflow to be created. tx.send(()).unwrap(); sync_queue .update(&mut app, |queue, ctx| { assert!(queue.is_dequeueing()); ctx.await_spawned_future(queue.spawned_futures[1]) }) .await; app.read(|ctx| { // The queue should still be dequeueing. assert!(sync_queue.as_ref(ctx).is_dequeueing()); let events = sync_queue_events.as_ref(ctx); assert_eq!(events.0.len(), 2); assert_eq!( &events.0, &[ SyncQueueEvent::ObjectCreationFailure { reason: CreationFailureReason::UniqueKeyConflict { id: gso_id.to_hash(), initiated_by: InitiatedBy::User }, }, SyncQueueEvent::ObjectCreationSuccessful { server_creation_info: ServerCreationInfo { server_id_and_type: ServerIdAndType { id: workflow_server_id, id_type: ObjectIdType::Workflow }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, client_id: workflow_id, revision_and_editor: RevisionAndLastEditor { revision: workflow_ts.into(), last_editor_uid: None }, metadata_ts: workflow_ts.into(), initiated_by: InitiatedBy::User } ] ); }); }); } #[test] fn test_dequeue_after_transient_failure() { App::test((), |mut app| async move { let owner = Owner::mock_current_user(); let success_notebook_id = ClientId::default(); let success_notebook_server_id = ServerId::from(1); let success_notebook_ts = DateTime::<Utc>::default(); let success_notebook = CloudNotebookModel { title: "Successful Notebook".to_string(), data: "Hello :)".to_string(), ai_document_id: None, conversation_id: None, }; let failure_notebook_id = ClientId::default(); let failure_notebook = CloudNotebookModel { title: "Failed Notebook".to_string(), data: "Hello :(".to_string(), ai_document_id: None, conversation_id: None, }; let (tx, rx) = std::sync::mpsc::channel(); let failure_attempts = 1 + super::DEFAULT_RETRY_OPTION.remaining_retries(); let mut cloud_objects_client_mock = MockObjectClient::new(); cloud_objects_client_mock .expect_create_notebook() .times(failure_attempts) .returning(move |_| Err(anyhow!("Transient network error!!!!"))); cloud_objects_client_mock .expect_create_notebook() .times(1) .return_once(move |_| { rx.recv().unwrap(); Ok(CreateCloudObjectResult::Success { created_cloud_object: CreatedCloudObject { client_id: success_notebook_id, revision_and_editor: RevisionAndLastEditor { revision: success_notebook_ts.into(), last_editor_uid: None, }, metadata_ts: success_notebook_ts.into(), server_id_and_type: ServerIdAndType { id: success_notebook_server_id, id_type: ObjectIdType::Notebook, }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, }) }); initialize_app(&mut app); let sync_queue = create_sync_queue( &mut app, vec![ QueueItem::CreateObject { object_type: ObjectType::Notebook, owner, id: failure_notebook_id, title: None, serialized_model: Some(Arc::new(failure_notebook.serialized())), initial_folder_id: None, entrypoint: CloudObjectEventEntrypoint::Unknown, initiated_by: InitiatedBy::User, }, QueueItem::CreateObject { object_type: ObjectType::Notebook, owner, id: success_notebook_id, title: None, serialized_model: Some(Arc::new(success_notebook.serialized())), initial_folder_id: None, entrypoint: CloudObjectEventEntrypoint::Unknown, initiated_by: InitiatedBy::User, }, ], cloud_objects_client_mock, false, ); let sync_queue_events = app.add_model(|_ctx| Events::default()); sync_queue_events.update(&mut app, |_, ctx| { ctx.subscribe_to_model(&sync_queue, |me, _, event, _ctx| me.0.push(event.clone())) }); sync_queue .update(&mut app, |queue, ctx| { queue.start_dequeueing(ctx); ctx.await_spawned_future(queue.spawned_futures[0]) }) .await; // Wait for the first notebook's creation to fail. // The failures are retried, but their futures are spawned on background threads, // so we can't access them. Instead, we wait for a SyncQueue event to appear in the model. assert_eventually!( 4000 => sync_queue_events.read(&app, |events, _ctx| !events.0.is_empty()), "Timed out waiting for failure" ); sync_queue_events.read(&app, |events, _| { assert_eq!( &events.0[0], &SyncQueueEvent::ObjectCreationFailure { reason: CreationFailureReason::Other { id: failure_notebook_id.to_hash(), initiated_by: InitiatedBy::User }, } ); }); // Allow the second notebook to be created. tx.send(()).unwrap(); sync_queue .update(&mut app, |queue, ctx| { assert!(queue.is_dequeueing()); ctx.await_spawned_future(queue.spawned_futures[1]) }) .await; sync_queue_events.read(&app, |events, _| { assert_eq!(events.0.len(), 2); assert_eq!( &events.0, &[ SyncQueueEvent::ObjectCreationFailure { reason: CreationFailureReason::Other { id: failure_notebook_id.to_hash(), initiated_by: InitiatedBy::User }, }, SyncQueueEvent::ObjectCreationSuccessful { server_creation_info: ServerCreationInfo { server_id_and_type: ServerIdAndType { id: success_notebook_server_id, id_type: ObjectIdType::Notebook }, creator_uid: None, permissions: ServerPermissions::mock_personal(), }, client_id: success_notebook_id, revision_and_editor: RevisionAndLastEditor { revision: success_notebook_ts.into(), last_editor_uid: None }, metadata_ts: success_notebook_ts.into(), initiated_by: InitiatedBy::User } ] ); }); }); }