diff --git a/components/ads-client/src/ads_store.rs b/components/ads-client/src/ads_store.rs index eb8c377d94..17ecf9fa89 100644 --- a/components/ads-client/src/ads_store.rs +++ b/components/ads-client/src/ads_store.rs @@ -7,9 +7,12 @@ use serde::{Deserialize, Serialize}; use crate::{ ads_store::{builder::AdsStoreBuilder, store::AdsStoreHolder}, common::bytesize::ByteSize, - mars::ad_response::{AdImage, AdSpoc, AdTile}, + mars::{ + ad_response::{AdImage, AdSpoc, AdTile}, + error::FetchAdsError, + }, }; -use std::path::Path; +use std::{collections::HashMap, path::Path}; /// Identification of placement sent and returned from MARS (eg: `mock_spoc_1`) #[derive(Debug, Hash, PartialEq, Eq, Clone)] @@ -19,9 +22,6 @@ impl PlacementId { pub fn new(s: &str) -> PlacementId { PlacementId(s.to_string()) } - pub fn into_inner(self) -> String { - self.0 - } } impl AsRef for PlacementId { @@ -30,13 +30,51 @@ impl AsRef for PlacementId { } } +impl From for PlacementId { + fn from(value: String) -> Self { + PlacementId(value) + } +} + +impl From for String { + fn from(value: PlacementId) -> Self { + value.0 + } +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub enum StorableAd { Image(AdImage), - Spoc(AdSpoc), + Spoc(Vec), Tile(AdTile), } +impl StorableAd { + pub fn into_image(self) -> Option { + if let StorableAd::Image(image) = self { + Some(image) + } else { + None + } + } + + pub fn into_spocs(self) -> Option> { + if let StorableAd::Spoc(spocs) = self { + Some(spocs) + } else { + None + } + } + + pub fn into_tile(self) -> Option { + if let StorableAd::Tile(tile) = self { + Some(tile) + } else { + None + } + } +} + pub struct AdsStore { holder: AdsStoreHolder, #[allow(dead_code)] @@ -61,6 +99,17 @@ impl AdsStore { self.holder.invalidate_ad_by_id(placement_id)?; Ok(()) } + + pub fn lookup(&self, placement_id: &PlacementId) -> Result, FetchAdsError> { + self.holder.lookup(placement_id) + } + + pub fn store_ads(&self, ads: HashMap) -> Result<(), FetchAdsError> { + for (placement_id, ad) in ads { + self.holder.store_ad(&placement_id, ad)?; + } + Ok(()) + } } #[cfg(test)] diff --git a/components/ads-client/src/client.rs b/components/ads-client/src/client.rs index 29d8f7dd2d..8c453fedbe 100644 --- a/components/ads-client/src/client.rs +++ b/components/ads-client/src/client.rs @@ -4,11 +4,13 @@ */ #[cfg(feature = "stateful")] -use crate::ads_store::AdsStore; +use crate::ads_store::{AdsStore, PlacementId, StorableAd}; use crate::common::bytesize::ByteSize; use crate::http_cache::{CachePolicy, HttpCache}; use crate::mars::ad_request::{AdPlacementRequest, AdRequestFlags}; use crate::mars::ad_response::{AdImage, AdResponse, AdResponseValue, AdSpoc, AdTile}; +#[cfg(feature = "stateful")] +use crate::mars::error::FetchAdsError; use crate::mars::error::{RecordClickError, RecordImpressionError, ReportAdError}; use crate::mars::{MARSClient, ReportReason}; #[cfg(feature = "stateful")] @@ -111,6 +113,71 @@ where self.client.clear_cache() } + #[cfg(feature = "stateful")] + pub fn store_ads( + &mut self, + ads: HashMap, + ) -> Result<(), FetchAdsError> { + let ads_store = self.ads_store.lock(); + if let Some(ads_store) = ads_store.as_ref() { + ads_store.store_ads(ads)?; + Ok(()) + } else { + Err(FetchAdsError::SqliteShutdown) + } + } + + #[cfg(feature = "stateful")] + pub fn get_stored_ad_images(&self, placement_id: &PlacementId) -> Option { + let ads_store = self.ads_store.lock(); + if let Some(ads_store) = ads_store.as_ref() { + match ads_store.lookup(placement_id) { + Ok(ad) => ad.and_then(|ad| ad.into_image()), + Err(_) => { + // TODO: Telemetry should return an error here (eg: some internal sqlite error) + None + } + } + } else { + // TODO: Telemetry should be added here for the database being shut down. + None + } + } + + #[cfg(feature = "stateful")] + pub fn get_stored_ad_spocs(&self, placement_id: &PlacementId) -> Option> { + let ads_store = self.ads_store.lock(); + if let Some(ads_store) = ads_store.as_ref() { + match ads_store.lookup(placement_id) { + Ok(ad) => ad.and_then(|ad| ad.into_spocs()), + Err(_) => { + // TODO: Telemetry should return an error here (eg: some internal sqlite error) + None + } + } + } else { + // TODO: Telemetry should be added here for the database being shut down. + None + } + } + + #[cfg(feature = "stateful")] + pub fn get_stored_ad_tile(&self, placement_id: &PlacementId) -> Option { + let ads_store = self.ads_store.lock(); + if let Some(ads_store) = ads_store.as_ref() { + match ads_store.lookup(placement_id) { + Ok(ad) => ad.and_then(|ad| ad.into_tile()), + Err(_) => { + // TODO: Telemetry should return an error here (eg: some internal sqlite error) + None + } + } + } else { + // TODO: Telemetry should be added here for the database being shut down. + None + } + } + pub fn get_context_id(&self) -> context_id::ApiResult { self.context_id_component.request(DEFAULT_ROTATION_DAYS) } diff --git a/components/ads-client/src/client/error.rs b/components/ads-client/src/client/error.rs index 2542939493..e6e648ce9c 100644 --- a/components/ads-client/src/client/error.rs +++ b/components/ads-client/src/client/error.rs @@ -3,10 +3,19 @@ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */ +#[cfg(feature = "stateful")] +use std::sync::mpsc::{RecvTimeoutError, TrySendError}; + use crate::mars::error::{FetchAdsError, RecordClickError, RecordImpressionError, ReportAdError}; +#[cfg(feature = "stateful")] +use crate::worker::command; #[derive(Debug, thiserror::Error)] pub enum ComponentError { + #[cfg(feature = "stateful")] + #[error("Error requesting ads from worker: {0}")] + BackgroundWorker(#[from] BackgroundWorkerError), + #[error("Error recording a click for a placement: {0}")] RecordClick(#[from] RecordClickError), @@ -28,3 +37,26 @@ pub enum RequestAdsError { #[error("Error requesting ads from MARS: {0}")] FetchAds(#[from] FetchAdsError), } + +#[cfg(feature = "stateful")] +#[derive(Debug, thiserror::Error)] +pub enum BackgroundWorkerError { + #[error("Error requesting new ads from the background worker: worker closed")] + Closed, + + #[error("Error requesting new ads from the background worker: worker full")] + Full, + + #[error("Background worker timed out waiting for response: {0}")] + TimedOut(#[from] RecvTimeoutError), +} + +#[cfg(feature = "stateful")] +impl From> for BackgroundWorkerError { + fn from(value: TrySendError) -> Self { + match value { + TrySendError::Disconnected(_) => BackgroundWorkerError::Closed, + TrySendError::Full(_) => BackgroundWorkerError::Full, + } + } +} diff --git a/components/ads-client/src/ffi.rs b/components/ads-client/src/ffi.rs index 6ea775bbab..e75e1ff574 100644 --- a/components/ads-client/src/ffi.rs +++ b/components/ads-client/src/ffi.rs @@ -20,6 +20,8 @@ use crate::mars::ad_response::{ }; use crate::mars::Environment; use crate::mars::ReportReason; +#[cfg(feature = "stateful")] +use crate::worker; use crate::AdsClientUrl; use crate::MozAdsClient; use parking_lot::Mutex; @@ -107,6 +109,13 @@ impl MozAdsClientBuilder { .take() .map(MozAdsTelemetryWrapper::new) .unwrap_or_else(MozAdsTelemetryWrapper::noop); + #[cfg(feature = "stateful")] + let store_set = inner.store_config.is_some(); + #[cfg(feature = "stateful")] + let worker_buffer_size = inner + .store_config + .as_ref() + .and_then(|x| x.worker_buffer_size); let client_config = AdsClientConfig { cache_config: inner.cache_config.clone().map(Into::into), environment: inner.environment.clone().unwrap_or_default().into(), @@ -116,9 +125,18 @@ impl MozAdsClientBuilder { }; let client = AdsClient::new(client_config); let shutdown_references = client.shutdown_references(); + let inner = Arc::new(Mutex::new(client)); + #[cfg(feature = "stateful")] + let worker = if store_set { + worker::BackgroundWorker::new(inner.clone(), worker_buffer_size) + } else { + worker::BackgroundWorker::new_empty() + }; MozAdsClient { - inner: Mutex::new(client), + inner, shutdown_references, + #[cfg(feature = "stateful")] + _worker: worker, } } @@ -172,6 +190,7 @@ pub struct MozAdsCacheConfig { #[derive(Clone, uniffi::Record)] pub struct MozAdsStoreConfig { pub db_path: String, + pub worker_buffer_size: Option, } #[derive(Debug, PartialEq, uniffi::Record)] diff --git a/components/ads-client/src/lib.rs b/components/ads-client/src/lib.rs index 6fbf581683..587fd3616a 100644 --- a/components/ads-client/src/lib.rs +++ b/components/ads-client/src/lib.rs @@ -3,7 +3,7 @@ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */ -use std::collections::HashMap; +use std::{collections::HashMap, sync::Arc}; use client::error::ComponentError; use error_support::handle_error; @@ -23,9 +23,13 @@ pub mod http_cache; mod mars; pub mod shutdown; pub mod telemetry; +#[cfg(feature = "stateful")] +pub mod worker; pub use ffi::*; +#[cfg(feature = "stateful")] +use crate::worker::BackgroundWorker; use crate::{ffi::telemetry::MozAdsTelemetryWrapper, shutdown::ShutdownReferences}; #[cfg(test)] @@ -39,10 +43,13 @@ uniffi::custom_type!(AdsClientUrl, String, { lower: |obj| obj.as_str().to_string(), }); +pub type MozAdsClientInner = Arc>>; #[derive(uniffi::Object)] pub struct MozAdsClient { - inner: Mutex>, + inner: MozAdsClientInner, shutdown_references: ShutdownReferences, + #[cfg(feature = "stateful")] + _worker: BackgroundWorker, } #[uniffi::export] diff --git a/components/ads-client/src/mars/error.rs b/components/ads-client/src/mars/error.rs index ce87833924..22f9f0573d 100644 --- a/components/ads-client/src/mars/error.rs +++ b/components/ads-client/src/mars/error.rs @@ -46,6 +46,9 @@ pub enum FetchAdsError { #[error("Internal database error: {0}")] Sqlite(#[from] rusqlite::Error), + #[error("Internal database error: database shut down or uninitialized")] + SqliteShutdown, + #[error("Error sending request: {0}")] Request(#[from] viaduct::ViaductError), diff --git a/components/ads-client/src/worker.rs b/components/ads-client/src/worker.rs new file mode 100644 index 0000000000..e679d572ca --- /dev/null +++ b/components/ads-client/src/worker.rs @@ -0,0 +1,90 @@ +use crate::{ + client::error::{BackgroundWorkerError, ComponentError}, + worker::command::DispatchCommand, + MozAdsClientInner, +}; +use std::{ + sync::mpsc::{self, Receiver, SyncSender}, + thread::JoinHandle, +}; + +pub mod command; + +// This is a somewhat arbitrary default value that is overridable. +pub const ADS_CLIENT_WORKER_CHANNEL_BUFFER_SIZE_DEFAULT: usize = 10000; +pub const ADS_CLIENT_WORKER_THREAD_NAME: &str = "ads-client.worker"; + +pub struct BackgroundWorker { + _worker_thread: Option>, + worker_dispatch: Option>, +} + +impl BackgroundWorker { + pub fn new( + inner_client: MozAdsClientInner, + worker_buffer_size: Option, + ) -> BackgroundWorker { + let worker_buffer_size = worker_buffer_size.and_then(|x| usize::try_from(x).ok()); + + let (tx, rx) = mpsc::sync_channel( + worker_buffer_size.unwrap_or(ADS_CLIENT_WORKER_CHANNEL_BUFFER_SIZE_DEFAULT), + ); + let Some(worker_thread) = std::thread::Builder::new() + .name(ADS_CLIENT_WORKER_THREAD_NAME.to_string()) + .spawn(move || crate::worker::worker(inner_client, rx)).inspect_err(|err| { + error_support::error!("Failed to create ads-client worker thread `{ADS_CLIENT_WORKER_THREAD_NAME}` with: {err}") + }).ok() + else { + return BackgroundWorker { _worker_thread: None, worker_dispatch: None } + }; + + BackgroundWorker { + _worker_thread: Some(worker_thread), + worker_dispatch: Some(tx), + } + } + + pub fn new_empty() -> BackgroundWorker { + BackgroundWorker { + _worker_thread: None, + worker_dispatch: None, + } + } + + pub fn dispatch(&self, command: DispatchCommand) -> Result<(), ComponentError> { + if let Some(worker_dispatch) = &self.worker_dispatch { + worker_dispatch + .try_send(command) + .map_err(BackgroundWorkerError::from)?; + + Ok(()) + } else { + Err(BackgroundWorkerError::Closed.into()) + } + } +} + +// Spawn worker thread from a reference to the client, returning a synchronous channel transmitter to the thread, and its JoinHandle. +// Returns None if thread fails to build. +pub fn build_worker_thread( + inner_client: MozAdsClientInner, + max_channel_size: Option, +) -> Option<(SyncSender, JoinHandle<()>)> { + let (tx, rx) = mpsc::sync_channel( + max_channel_size.unwrap_or(ADS_CLIENT_WORKER_CHANNEL_BUFFER_SIZE_DEFAULT), + ); + let worker_thread_handle = std::thread::Builder::new() + .name(ADS_CLIENT_WORKER_THREAD_NAME.to_string()) + .spawn(move || crate::worker::worker(inner_client, rx)).inspect_err(|err| { + error_support::error!("Failed to create ads-client worker thread `{ADS_CLIENT_WORKER_THREAD_NAME}` with: {err}") + }).ok()?; + Some((tx, worker_thread_handle)) +} + +fn worker(inner_client: MozAdsClientInner, rx: Receiver) { + // Synchronously run tasks in the order they are passed in this separate channel. + while let Ok(command) = rx.recv() { + // Error is naturally logged through `handle_error` conversion macro. + let _ = command.run_command(&inner_client); + } +} diff --git a/components/ads-client/src/worker/command.rs b/components/ads-client/src/worker/command.rs new file mode 100644 index 0000000000..33783a2d6d --- /dev/null +++ b/components/ads-client/src/worker/command.rs @@ -0,0 +1,140 @@ +use std::collections::HashMap; + +use error_support::handle_error; + +use crate::{ + ads_store::StorableAd, + client::error::{ComponentError, RequestAdsError}, + http_cache::CachePolicy, + mars::ad_request::AdPlacementRequest, + AdsClientApiResult, MozAdsClientInner, MozAdsPlacementRequest, MozAdsPlacementRequestWithCount, +}; + +// Command dispatch enum for passing different instructions to the background worker thread. +// `RequestImageAds`, `RequestSpocAds`, `RequestTileAds` are prefetch mechanisms that query and load data into the local cache. +pub enum DispatchCommand { + RequestImageAds { + image_ad_requests: Vec, + cache_policy: CachePolicy, + ohttp: bool, + flags: HashMap, + blocks: Vec, + }, + RequestSpocAds { + spoc_ad_requests: Vec, + cache_policy: CachePolicy, + ohttp: bool, + flags: HashMap, + blocks: Vec, + }, + RequestTileAds { + tile_ad_requests: Vec, + cache_policy: CachePolicy, + ohttp: bool, + flags: HashMap, + blocks: Vec, + }, +} + +impl DispatchCommand { + // Runs a dispatched command synchronously in it's thread. + // The dispatched command calls the corresponding `AdsClient` synchronous method, meaning that behavior between the two is shared. + #[handle_error(ComponentError)] + pub fn run_command(self, ads_client_inner: &MozAdsClientInner) -> AdsClientApiResult<()> { + match self { + DispatchCommand::RequestImageAds { + image_ad_requests, + cache_policy, + flags, + ohttp, + blocks, + } => { + let mut inner = ads_client_inner.lock(); + if !image_ad_requests.is_empty() { + let image_ad_requests: Vec = + image_ad_requests.iter().map(|r| r.into()).collect(); + let image_response = inner + .request_image_ads( + image_ad_requests, + flags, + Some(cache_policy), + ohttp, + blocks, + ) + .map_err(ComponentError::RequestAds)?; + inner + .store_ads( + image_response + .into_iter() + .map(|(k, v)| (k.into(), StorableAd::Image(v))) + .collect(), + ) + .map_err(RequestAdsError::from)?; + } + Ok(()) + } + DispatchCommand::RequestSpocAds { + spoc_ad_requests, + cache_policy, + flags, + ohttp, + blocks, + } => { + let mut inner = ads_client_inner.lock(); + if !spoc_ad_requests.is_empty() { + let spoc_ad_requests: Vec = + spoc_ad_requests.iter().map(|r| r.into()).collect(); + let spoc_response = inner + .request_spoc_ads( + spoc_ad_requests, + flags, + Some(cache_policy), + ohttp, + blocks, + ) + .map_err(ComponentError::RequestAds)?; + inner + .store_ads( + spoc_response + .into_iter() + .map(|(k, v)| (k.into(), StorableAd::Spoc(v))) + .collect(), + ) + .map_err(RequestAdsError::from)?; + } + Ok(()) + } + DispatchCommand::RequestTileAds { + tile_ad_requests, + cache_policy, + flags, + ohttp, + blocks, + } => { + let mut inner = ads_client_inner.lock(); + if !tile_ad_requests.is_empty() { + let tile_ad_requests: Vec = + tile_ad_requests.iter().map(|r| r.into()).collect(); + let tile_response = inner + .request_tile_ads( + tile_ad_requests, + flags, + Some(cache_policy), + ohttp, + blocks, + ) + .map_err(ComponentError::RequestAds)?; + inner + .store_ads( + tile_response + .into_iter() + .map(|(k, v)| (k.into(), StorableAd::Tile(v))) + .collect(), + ) + .map_err(RequestAdsError::from)?; + } + Ok(()) + } + } + } +}