diff --git a/CHANGELOG.md b/CHANGELOG.md index e287c44668..638b00c7c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # Unreleased changes +* Add mechanisms for storing and retrieving submitted pings ([#3585](https://github.com/mozilla/glean/pull/3585)). + * Add new `submitted_pings` table to the SQLite database. + * Add methods to store, retrieve, update, and clear stored submitted pings. + * Update Ping and uploader implementations to store and update submitted pings as appropriate. + [Full changelog](https://github.com/mozilla/glean/compare/v70.0.0...main) # v70.0.0 (2026-08-20) @@ -22,8 +27,8 @@ * Implement glean-noop as a feature of glean-sym ([#3541](https://github.com/mozilla/glean/pull/3541)) * Support pings ([#3544](https://github.com/mozilla/glean/pull/3544)) * Implement the event metric ([#3534](https://github.com/mozilla/glean/pull/3534)) - * BREAKING CHANGE: Switch from a noop feature to an `active` feature ([#3583](https://github.com/mozilla/glean/pull/3583)) * iOS + * BREAKING CHANGE: Switch from a noop feature to an `active` feature ([#3583](https://github.com/mozilla/glean/pull/3583)) * Implement the custom distribution metric type ([#3572](https://github.com/mozilla/glean/pull/3572)) * Python * Implement the custom distribution metric type ([#3572](https://github.com/mozilla/glean/pull/3572)) diff --git a/glean-core/benchmark/benches/dispatcher.rs b/glean-core/benchmark/benches/dispatcher.rs index f23881afbd..4f82c675c8 100644 --- a/glean-core/benchmark/benches/dispatcher.rs +++ b/glean-core/benchmark/benches/dispatcher.rs @@ -89,6 +89,7 @@ pub fn metric_dispatcher_benchmark(c: &mut Criterion) { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }; let client_info = ClientInfoMetrics::unknown(); diff --git a/glean-core/benchmark/benches/lifetime_buffering.rs b/glean-core/benchmark/benches/lifetime_buffering.rs index cd31f80b95..76c4208bd7 100644 --- a/glean-core/benchmark/benches/lifetime_buffering.rs +++ b/glean-core/benchmark/benches/lifetime_buffering.rs @@ -4,7 +4,7 @@ //! Benchmark the impact of `delay_ping_lifetime_io` and automatic flushing on the overall performance. -use criterion::{Criterion, criterion_group, criterion_main}; +use criterion::{criterion_group, criterion_main, Criterion}; use glean_core::{CommonMetricData, CounterMetric, Glean, Lifetime}; pub fn delay_io_benchmark(c: &mut Criterion) { @@ -37,6 +37,7 @@ pub fn delay_io_benchmark(c: &mut Criterion) { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }; let glean = Glean::new(cfg).unwrap(); @@ -85,6 +86,7 @@ pub fn delay_io_benchmark(c: &mut Criterion) { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }; let glean = Glean::new(cfg).unwrap(); @@ -133,6 +135,7 @@ pub fn delay_io_benchmark(c: &mut Criterion) { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }; let glean = Glean::new(cfg).unwrap(); diff --git a/glean-core/examples/rkv-open.rs b/glean-core/examples/rkv-open.rs index 3e72ac0d1e..e97a4e1c01 100644 --- a/glean-core/examples/rkv-open.rs +++ b/glean-core/examples/rkv-open.rs @@ -57,6 +57,7 @@ fn main() { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }; let client_info = ClientInfoMetrics::unknown(); diff --git a/glean-core/rlb/src/configuration.rs b/glean-core/rlb/src/configuration.rs index fd10141b94..85e74bfa0d 100644 --- a/glean-core/rlb/src/configuration.rs +++ b/glean-core/rlb/src/configuration.rs @@ -67,6 +67,8 @@ pub struct Configuration { pub session_inactivity_timeout: Duration, /// The number of "events" pings to accelerate each session, plus one. pub events_ping_acceleration_factor: Option, + /// Whether to store submitted pings or not + pub enable_store_submitted_pings: bool, } /// Configuration builder. @@ -131,6 +133,8 @@ pub struct Builder { pub session_inactivity_timeout: Duration, /// The number of "events" pings to accelerate each session, plus one. pub events_ping_acceleration_factor: Option, + /// Whether to store submitted pings or not. + pub enable_store_submitted_pings: bool, } impl Builder { @@ -162,6 +166,7 @@ impl Builder { session_sample_rate: 1.0, session_inactivity_timeout: Duration::from_secs(30 * 60), events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, } } @@ -189,6 +194,7 @@ impl Builder { session_sample_rate: self.session_sample_rate, session_inactivity_timeout: self.session_inactivity_timeout, events_ping_acceleration_factor: self.events_ping_acceleration_factor, + enable_store_submitted_pings: self.enable_store_submitted_pings, } } @@ -293,4 +299,10 @@ impl Builder { self.events_ping_acceleration_factor = Some(factor); self } + + /// Set whether to store submitted pings or not. + pub fn with_store_submitted_pings_enabled(mut self, value: bool) -> Self { + self.enable_store_submitted_pings = value; + self + } } diff --git a/glean-core/rlb/src/lib.rs b/glean-core/rlb/src/lib.rs index 17ddd84d3b..63feda7e9b 100644 --- a/glean-core/rlb/src/lib.rs +++ b/glean-core/rlb/src/lib.rs @@ -136,6 +136,7 @@ fn initialize_internal(cfg: Configuration, client_info: ClientInfoMetrics) -> Op session_sample_rate: cfg.session_sample_rate, session_inactivity_timeout_ms: cfg.session_inactivity_timeout.as_millis() as u64, events_ping_acceleration_factor: cfg.events_ping_acceleration_factor.map(|x| x as u32), + enable_store_submitted_pings: cfg.enable_store_submitted_pings, }; glean_core::glean_initialize(core_cfg, client_info.into(), callbacks); @@ -180,6 +181,32 @@ pub fn set_collection_enabled(enabled: bool) { glean_core::glean_set_collection_enabled(enabled) } +/// Sets whether storing submitted pings is enabled or not. +pub fn set_store_submitted_pings_enabled(enabled: bool) { + glean_core::glean_set_store_submitted_pings_enabled(enabled) +} + +/// Returns all stored submitted pings. +/// +/// Requires storing submitted pings to be enabled. +/// See [`set_store_submitted_pings_enabled`]. +pub fn get_all_stored_submitted_pings() -> Vec { + glean_core::glean_get_all_stored_submitted_pings() +} + +/// Returns all stored submitted pings with a given ping name. +/// +/// Requires storing submitted pings to be enabled. +/// See [`set_store_submitted_pings_enabled`]. +pub fn get_stored_submitted_pings_by_name(ping: String) -> Vec { + glean_core::glean_get_stored_submitted_pings_by_name(ping) +} + +/// Clears all stored submitted pings. +pub fn clear_stored_submitted_pings() { + glean_core::glean_clear_stored_submitted_pings() +} + /// Collects and submits a ping for eventual uploading by name. /// /// Note that this needs to be public in order for RLB consumers to diff --git a/glean-core/src/core/mod.rs b/glean-core/src/core/mod.rs index 397625dd00..0d250528ae 100644 --- a/glean-core/src/core/mod.rs +++ b/glean-core/src/core/mod.rs @@ -146,6 +146,7 @@ where /// session_sample_rate: 1.0, /// session_inactivity_timeout_ms: 1_800_000, /// events_ping_acceleration_factor: None, +/// enable_store_submitted_pings: false, /// }; /// let mut glean = Glean::new(cfg).unwrap(); /// let ping = PingType::new("sample", true, false, true, true, true, vec![], vec![], true, vec![]); @@ -196,6 +197,7 @@ pub struct Glean { #[ignore_malloc_size_of = "TODO: Expose session memory allocations (bug 2043355)"] pub(crate) session_manager: SessionManager, events_ping_acceleration_factor: Option, + pub(crate) store_submitted_pings_enabled: bool, } impl Glean { @@ -279,6 +281,7 @@ impl Glean { events_ping_acceleration_factor: cfg .events_ping_acceleration_factor .map(|x| x as usize), + store_submitted_pings_enabled: cfg.enable_store_submitted_pings, }; // Ensuring these pings are registered. @@ -604,6 +607,7 @@ impl Glean { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }; let mut glean = Self::new(cfg).unwrap(); @@ -817,6 +821,16 @@ impl Glean { } } + /// Sets whether storing submitted pings is enabled or not. + /// + /// # Arguments + /// + /// * `enabled` - When true, enables storing submitted pings. + /// + pub fn set_store_submitted_pings_enabled(&mut self, enabled: bool) { + self.store_submitted_pings_enabled = enabled; + } + /// Enable or disable a ping. /// /// Disabling a ping causes all data for that ping to be removed from storage diff --git a/glean-core/src/database/sqlite.rs b/glean-core/src/database/sqlite.rs index 1db1eeb2d1..d055dcd367 100644 --- a/glean-core/src/database/sqlite.rs +++ b/glean-core/src/database/sqlite.rs @@ -9,14 +9,15 @@ use std::path::Path; use std::str; use std::time::Duration; +use chrono::{DateTime, Utc}; +use connection::Connection; use malloc_size_of::MallocSizeOf; -use rusqlite::params; -use rusqlite::types::FromSqlError; +use rusqlite::fallible_iterator::FallibleIterator; +use rusqlite::types::{FromSql, FromSqlError, FromSqlResult, ToSqlOutput, ValueRef}; use rusqlite::OptionalExtension; use rusqlite::Transaction; +use rusqlite::{params, ToSql}; use rusqlite::{Error as SqlError, ErrorCode}; - -use connection::Connection; use schema::Schema; pub use schema::SchemaError; @@ -24,9 +25,9 @@ use crate::common_metric_data::CommonMetricDataInternal; use crate::database::migration::{self, MigrationState}; use crate::metrics::dual_labeled_counter::RECORD_SEPARATOR; use crate::metrics::Metric; -use crate::Glean; use crate::Lifetime; use crate::Result; +use crate::{Glean, JsonValue}; use super::ConnExt; @@ -72,6 +73,48 @@ impl MallocSizeOf for Database { } } +pub struct SubmittedPing { + pub document_id: String, + pub ping: String, + pub submitted_date: SqliteDatetime, + pub uploaded_date: Option, + pub upload_failed: bool, + pub payload: Option, +} + +impl SubmittedPing { + pub fn payload(&self) -> Option { + self.payload + .as_ref() + .map(|p| match serde_json::from_str(p) { + Ok(v) => Some(v), + Err(e) => { + log::warn!("Unable to serialize JSON payload from string: {:?}", e); + None + } + }) + .unwrap_or(None) + } +} + +#[derive(Debug, Clone, PartialEq)] +pub struct SqliteDatetime(pub DateTime); + +impl ToSql for SqliteDatetime { + fn to_sql(&self) -> rusqlite::Result> { + Ok(ToSqlOutput::from(self.0.timestamp_millis())) + } +} + +impl FromSql for SqliteDatetime { + fn column_result(value: ValueRef<'_>) -> FromSqlResult { + i64::column_result(value).and_then(|as_i64| match DateTime::from_timestamp_millis(as_i64) { + Some(d) => Ok(SqliteDatetime(d)), + None => Err(FromSqlError::InvalidType), + }) + } +} + const DEFAULT_DATABASE_FILE_NAME: &str = "glean.sqlite"; /// Calculate the database size from all the files in the directory. @@ -448,6 +491,193 @@ impl Database { .unwrap_or(false) } + /// Gets all pings in the `submitted_pings` table. + pub fn get_all_submitted_pings(&self) -> Vec { + let get_all_submitted_pings_sql = r#" + SELECT + document_id, + ping, + date_submitted, + date_uploaded, + upload_failed, + payload + FROM submitted_pings + ORDER BY date_submitted DESC + "#; + self.conn + .read(|conn| { + let Ok(mut stmt) = conn.prepare_cached(get_all_submitted_pings_sql) else { + return Ok(Default::default()); + }; + let Ok(pings_iter) = stmt.query([]) else { + return Ok(Default::default()); + }; + + pings_iter + .map(|r| { + Ok(SubmittedPing { + document_id: r.get(0).unwrap(), + ping: r.get(1).unwrap(), + submitted_date: r.get(2).unwrap(), + uploaded_date: r.get(3).unwrap(), + upload_failed: r.get(4).unwrap(), + payload: r.get(5).unwrap(), + }) + }) + .collect() + }) + .unwrap_or_default() + } + + /// Returns all submitted pings in the `submitted_pings` table that match a supplied ping name. + /// + /// # Arguments + /// + /// * `ping` - The name of the pings to return. + pub fn get_submitted_pings_by_name(&self, ping: &str) -> Vec { + let get_submitted_pings_sql = r#" + SELECT + document_id, + ping, + date_submitted, + date_uploaded, + upload_failed, + payload + FROM submitted_pings + WHERE + ping = ?1 + ORDER BY date_submitted DESC + "#; + self.conn + .read(|conn| { + let Ok(mut stmt) = conn.prepare_cached(get_submitted_pings_sql) else { + return Ok(Default::default()); + }; + let Ok(pings_iter) = stmt.query([ping]) else { + return Ok(Default::default()); + }; + + pings_iter + .map(|r| { + Ok(SubmittedPing { + document_id: r.get(0).unwrap(), + ping: r.get(1).unwrap(), + submitted_date: r.get(2).unwrap(), + uploaded_date: r.get(3).unwrap(), + upload_failed: r.get(4).unwrap(), + payload: r.get(5).unwrap(), + }) + }) + .collect() + }) + .unwrap_or_default() + } + + /// Marks a particular ping as uploaded. + /// + /// # Arguments + /// + /// * `document_id` - The ping to mark as uploaded. + /// * `date_uploaded` - The UTC date/time the ping was uploaded. + /// + /// # Returns + /// + /// A `usize` representing the number of rows updated. + pub fn mark_ping_as_uploaded(&self, document_id: &str, date_uploaded: DateTime) -> usize { + let update_submitted_pings_sql = + "UPDATE submitted_pings SET date_uploaded = ?1 WHERE document_id = ?2"; + self.conn + .write(|tx| { + let Ok(mut stmt) = tx.prepare_cached(update_submitted_pings_sql) else { + return Ok(Default::default()); + }; + stmt.execute(params![SqliteDatetime(date_uploaded), document_id]) + }) + .unwrap_or_default() + } + + pub fn mark_ping_as_upload_failed(&self, document_id: &str) -> usize { + let update_submitted_pings_sql = + "UPDATE submitted_pings SET upload_failed = 1 WHERE document_id = ?1"; + self.conn + .write(|tx| { + let Ok(mut stmt) = tx.prepare_cached(update_submitted_pings_sql) else { + return Ok(Default::default()); + }; + stmt.execute(params![document_id]) + }) + .unwrap_or_default() + } + + /// Stores a submitted ping into the `submitted_pings` table. + /// + /// # Arguments + /// + /// * `document_id` - The unique identifier for the ping. + /// * `ping` - The name of the ping. + /// * `date_submitted` - The UTC date/time the ping was submitted. + /// * `date_uploaded` - An optional UTC date/time the ping was uploaded. + /// * `payload` - A JSON representation of the content of the ping. + /// + /// # Returns + /// + /// An empty `Result`. + pub fn store_submitted_ping( + &self, + document_id: &str, + ping: &str, + date_submitted: DateTime, + date_uploaded: Option>, + upload_failed: bool, + payload: JsonValue, + ) -> Result<()> { + self.conn.write(|tx| { + let insert_sql = r#" + INSERT INTO + submitted_pings (document_id, ping, date_submitted, date_uploaded, upload_failed, payload) + VALUES + (?1, ?2, ?3, ?4, ?5, ?6) + ON CONFLICT(document_id) DO UPDATE SET + ping = excluded.ping, + date_submitted = excluded.date_submitted, + date_uploaded = excluded.date_uploaded, + upload_failed = excluded.upload_failed, + payload = excluded.payload + "#; + let mut stmt = tx.prepare_cached(insert_sql)?; + stmt.execute(params![ + document_id, + ping, + SqliteDatetime(date_submitted), + date_uploaded.map(SqliteDatetime), + upload_failed, + serde_json::to_string(&payload).expect("Unable to convert JSON payload to string.") + ])?; + Ok(()) + }) + } + + /// Remove stored submitted pings that are older than `before_time` (or 30 days if not specified) + /// + /// # Arguments + /// + /// * `before_time` - An optional date – when supplied uses that date as the oldest date_submitted we should keep. + /// Defaults to 30 days if `None` is supplied. + /// + /// # Returns + /// + /// An empty `Result`. + pub fn cleanup_submitted_pings(&self, before_time: Option>) -> Result<()> { + let days_30 = Duration::from_secs(30 * 24 * 60 * 60); + let time = before_time.unwrap_or_else(|| Utc::now() - days_30); + let delete_sql = "DELETE FROM submitted_pings WHERE date_submitted <= ?1"; + self.conn.write(|tx| { + let mut stmt = tx.prepare_cached(delete_sql)?; + stmt.execute(params![SqliteDatetime(time)]) + })?; + Ok(()) + } + /// Records a metric in the underlying storage system. pub fn record(&self, glean: &Glean, data: &CommonMetricDataInternal, value: &Metric) { let name = data.base_identifier(); diff --git a/glean-core/src/database/sqlite/schema.rs b/glean-core/src/database/sqlite/schema.rs index cec5072214..175016d1a9 100644 --- a/glean-core/src/database/sqlite/schema.rs +++ b/glean-core/src/database/sqlite/schema.rs @@ -16,7 +16,7 @@ use super::connection::ConnectionOpener; pub struct Schema; impl ConnectionOpener for Schema { - const MAX_SCHEMA_VERSION: u32 = 2; + const MAX_SCHEMA_VERSION: u32 = 3; type Error = SchemaError; @@ -53,15 +53,24 @@ impl ConnectionOpener for Schema { fn create(tx: &mut Transaction<'_>) -> Result<(), Self::Error> { tx.execute_batch( " - CREATE TABLE telemetry( - id TEXT NOT NULL, - ping TEXT NOT NULL, - lifetime TEXT NOT NULL, - labels TEXT NOT NULL, -- can't be null or ON CONFLICT won't work - value BLOB, - UNIQUE(id, ping, labels) - ); - CREATE TABLE migration(id INTEGER PRIMARY KEY, state TEXT NOT NULL); + CREATE TABLE telemetry( + id TEXT NOT NULL, + ping TEXT NOT NULL, + lifetime TEXT NOT NULL, + labels TEXT NOT NULL, -- can't be null or ON CONFLICT won't work + value BLOB, + UNIQUE(id, ping, labels) + ); + CREATE TABLE migration(id INTEGER PRIMARY KEY, state TEXT NOT NULL); + CREATE TABLE submitted_pings( + document_id TEXT PRIMARY KEY, + ping TEXT NOT NULL, + date_submitted INTEGER NOT NULL, + date_uploaded INTEGER, + upload_failed BOOLEAN NOT NULL, + payload TEXT + ); + CREATE INDEX submitted_pings_ping on submitted_pings(ping); ", )?; Ok(()) @@ -90,6 +99,24 @@ impl ConnectionOpener for Schema { } Ok(()) } + 3 => { + log::info!("Upgrading user_version to 3"); + // Clients upgrading to schema 3 don't have the table or index + tx.execute_batch( + " + CREATE TABLE submitted_pings( + document_id TEXT PRIMARY KEY, + ping TEXT NOT NULL, + date_submitted INTEGER NOT NULL, + date_uploaded INTEGER, + upload_failed BOOLEAN NOT NULL, + payload TEXT + ); + CREATE INDEX submitted_pings_ping on submitted_pings(ping); + ", + )?; + Ok(()) + } to_version => Err(SchemaError::UnsupportedSchemaVersion(to_version)), } } diff --git a/glean-core/src/glean.udl b/glean-core/src/glean.udl index 128393d09e..eea5e205f2 100644 --- a/glean-core/src/glean.udl +++ b/glean-core/src/glean.udl @@ -27,6 +27,14 @@ namespace glean { void glean_set_upload_enabled(boolean enabled); + void glean_set_store_submitted_pings_enabled(boolean enabled); + + sequence glean_get_all_stored_submitted_pings(); + + sequence glean_get_stored_submitted_pings_by_name(string ping); + + void glean_clear_stored_submitted_pings(); + // Experiment reporting API void glean_set_experiment_active(string experiment_id, string branch, record extra); void glean_set_experiment_inactive(string experiment_id); @@ -95,6 +103,15 @@ namespace glean { DistributionMetrics glean_test_get_distribution(); }; +dictionary SubmittedPing { + string document_id; + string ping; + string submitted_date; + string? uploaded_date; + boolean upload_failed; + JsonValue? payload; +}; + // A `Cow<'static, str>`, but really it's always the owned part. [Custom] typedef string CowString; @@ -127,6 +144,7 @@ dictionary InternalConfiguration { f64 session_sample_rate; // Must be in [0.0, 1.0]; values outside are clamped. u64 session_inactivity_timeout_ms; // Milliseconds; 0 means sessions never time out. u32? events_ping_acceleration_factor; + boolean enable_store_submitted_pings = false; }; // Session management mode. diff --git a/glean-core/src/lib.rs b/glean-core/src/lib.rs index 23c156ce0d..1d72fb21be 100644 --- a/glean-core/src/lib.rs +++ b/glean-core/src/lib.rs @@ -184,6 +184,8 @@ pub struct InternalConfiguration { pub session_inactivity_timeout_ms: u64, /// The number of "events" pings to accelerate each session, plus one. pub events_ping_acceleration_factor: Option, + /// Whether to store submitted pings. Default: false + pub enable_store_submitted_pings: bool, } /// How to specify the rate at which pings may be uploaded before they are throttled. @@ -795,6 +797,9 @@ pub fn shutdown() { } if let Some(database) = &glean.data_store { + if let Err(e) = database.cleanup_submitted_pings(None) { + log::info!("Could not clean up submitted_pings table: {:?}", e); + } if let Err(e) = database.run_maintenance(false) { log::info!("Can't run database maintenance on shutdown: {:?}", e); } @@ -965,6 +970,78 @@ pub fn glean_set_collection_enabled(enabled: bool) { glean_set_upload_enabled(enabled) } +/// Sets whether Glean should store submitted pings or not. +pub fn glean_set_store_submitted_pings_enabled(enabled: bool) { + if !was_initialize_called() { + return; + } + + launch_with_glean_mut(move |glean| { + glean.store_submitted_pings_enabled = enabled; + }); +} + +/// A submitted ping that has been stored by Glean. +pub struct SubmittedPing { + /// The document ID (unique identifier) + document_id: String, + /// The ping's name + ping: String, + /// RFC3339 datetime string + submitted_date: String, + /// Optional RFC3339 datetime string + uploaded_date: Option, + /// Whether the upload failed unrecoverably or not + upload_failed: bool, + /// The ping's payload + payload: Option, +} + +impl From for SubmittedPing { + fn from(value: database::sqlite::SubmittedPing) -> Self { + SubmittedPing { + document_id: value.document_id.clone(), + ping: value.ping.clone(), + submitted_date: value.submitted_date.0.to_rfc3339(), + uploaded_date: value.uploaded_date.as_ref().map(|d| d.0.to_rfc3339()), + upload_failed: value.upload_failed, + payload: value.payload(), + } + } +} + +/// Returns a `Vec` containing all stored submitted pings. +pub fn glean_get_all_stored_submitted_pings() -> Vec { + core::with_glean(|glean| glean.storage().get_all_submitted_pings()) + .into_iter() + .map(|p| p.into()) + .collect() +} + +/// Returns a `Vec` containing all stored submitted pings with the supplied name. +/// +/// # Arguments +/// +/// * `ping` - The name of the pings that should be returned. +pub fn glean_get_stored_submitted_pings_by_name(ping: String) -> Vec { + core::with_glean(|glean| glean.storage().get_submitted_pings_by_name(&ping)) + .into_iter() + .map(|p| p.into()) + .collect() +} + +/// Clears the stored submitted pings. +pub fn glean_clear_stored_submitted_pings() { + launch_with_glean(|glean| { + if let Err(e) = glean + .storage() + .cleanup_submitted_pings(Some(chrono::Utc::now())) + { + log::warn!("Unable to clear stored submitted pings: {:?}", e); + } + }); +} + /// Enable or disable a ping. /// /// Disabling a ping causes all data for that ping to be removed from storage diff --git a/glean-core/src/lib_unit_tests.rs b/glean-core/src/lib_unit_tests.rs index 55bab5f58d..f85a431cef 100644 --- a/glean-core/src/lib_unit_tests.rs +++ b/glean-core/src/lib_unit_tests.rs @@ -238,6 +238,7 @@ fn experimentation_id_is_set_correctly() { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }) .unwrap(); diff --git a/glean-core/src/metrics/ping.rs b/glean-core/src/metrics/ping.rs index fff04ddaa1..e10a0ddb7d 100644 --- a/glean-core/src/metrics/ping.rs +++ b/glean-core/src/metrics/ping.rs @@ -2,13 +2,13 @@ // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at https://mozilla.org/MPL/2.0/. -use std::fmt; -use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; - use crate::ping::PingMaker; use crate::upload::PingPayload; use crate::Glean; +use chrono::Utc; +use std::fmt; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; use malloc_size_of_derive::MallocSizeOf; use uuid::Uuid; @@ -353,6 +353,19 @@ impl PingType { .add_sync(glean, 1); } + if glean.store_submitted_pings_enabled { + if let Err(e) = glean.storage().store_submitted_ping( + ping.doc_id, + &self.0.name, + Utc::now(), + None, + false, + ping.content.clone(), + ) { + log::warn!("Storing a submitted ping failed: {e}"); + } + } + if let Err(e) = ping_maker.store_ping(glean.get_data_path(), &ping) { log::warn!( "IO error while writing ping to file: {}. Enqueuing upload of what we have in memory.", diff --git a/glean-core/src/upload/mod.rs b/glean-core/src/upload/mod.rs index 889ab3d1c2..7199a5848e 100644 --- a/glean-core/src/upload/mod.rs +++ b/glean-core/src/upload/mod.rs @@ -799,6 +799,11 @@ impl PingUploadManager { .set_stop_and_accumulate(glean, success_id, stop_time); self.upload_metrics.send_failure.cancel_sync(failure_id); } + if glean.store_submitted_pings_enabled { + glean + .storage() + .mark_ping_as_uploaded(document_id, Utc::now()); + } self.directory_manager.delete_file(document_id); } @@ -814,6 +819,9 @@ impl PingUploadManager { .send_failure .set_stop_and_accumulate(glean, failure_id, stop_time); } + if glean.store_submitted_pings_enabled { + glean.storage().mark_ping_as_upload_failed(document_id); + } self.directory_manager.delete_file(document_id); } @@ -2128,4 +2136,103 @@ mod test { UploadResult::http_status(200), ); } + + #[test] + fn stores_pings_during_submission_and_upload_if_enabled() { + let (mut glean, _t) = new_glean(None); + glean.set_store_submitted_pings_enabled(true); + + // Register a ping for testing + let ping_type = PingType::new( + "test", + true, + /* send_if_empty */ true, + true, + true, + true, + vec![], + vec![], + true, + vec![], + ); + glean.register_ping_type(&ping_type); + + // Submit a ping + ping_type.submit_sync(&glean, None); + + let pings = glean.storage().get_all_submitted_pings(); + assert_eq!(pings.len(), 1); + let ping = pings.first().unwrap(); + assert!(ping.submitted_date.0 <= Utc::now()); + assert!(ping.uploaded_date.is_none()); + + // Get the submitted PingRequest + match glean.get_upload_task() { + PingUploadTask::Upload { request } => { + // Simulate the processing of a sucessful request + let document_id = request.document_id; + glean.process_ping_upload_response(&document_id, UploadResult::http_status(200)); + } + _ => panic!("Expected upload manager to return the next request!"), + } + + let pings = glean.storage().get_all_submitted_pings(); + assert_eq!(pings.len(), 1); + let ping = pings.first().unwrap(); + assert!(ping.submitted_date.0 <= Utc::now()); + assert!(ping.uploaded_date.is_some()); + + // Verify that after request is returned, none are left + assert_eq!(glean.get_upload_task(), PingUploadTask::done()); + } + + #[test] + fn stores_pings_during_submission_and_marks_as_upload_failed_when_appropriate() { + let (mut glean, _t) = new_glean(None); + glean.set_store_submitted_pings_enabled(true); + + // Register a ping for testing + let ping_type = PingType::new( + "test", + true, + /* send_if_empty */ true, + true, + true, + true, + vec![], + vec![], + true, + vec![], + ); + glean.register_ping_type(&ping_type); + + // Submit a ping + ping_type.submit_sync(&glean, None); + + let pings = glean.storage().get_all_submitted_pings(); + assert_eq!(pings.len(), 1); + let ping = pings.first().unwrap(); + assert!(ping.submitted_date.0 <= Utc::now()); + assert!(ping.uploaded_date.is_none()); + + // Get the submitted PingRequest + match glean.get_upload_task() { + PingUploadTask::Upload { request } => { + // Simulate the processing of a sucessful request + let document_id = request.document_id; + glean.process_ping_upload_response(&document_id, UploadResult::http_status(400)); + } + _ => panic!("Expected upload manager to return the next request!"), + } + + let pings = glean.storage().get_all_submitted_pings(); + assert_eq!(pings.len(), 1); + let ping = pings.first().unwrap(); + assert!(ping.submitted_date.0 <= Utc::now()); + assert!(ping.upload_failed); + assert!(ping.uploaded_date.is_none()); + + // Verify that after request is returned, none are left + assert_eq!(glean.get_upload_task(), PingUploadTask::done()); + } } diff --git a/glean-core/tests/common/mod.rs b/glean-core/tests/common/mod.rs index b602beecb6..96e88f4fb2 100644 --- a/glean-core/tests/common/mod.rs +++ b/glean-core/tests/common/mod.rs @@ -75,6 +75,7 @@ pub fn new_glean_with_upload( session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }; let mut glean = Glean::new(cfg).unwrap(); diff --git a/glean-core/tests/event.rs b/glean-core/tests/event.rs index 0205a4a807..4b0183805a 100644 --- a/glean-core/tests/event.rs +++ b/glean-core/tests/event.rs @@ -559,6 +559,7 @@ fn with_event_timestamps() { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }; let mut glean = Glean::new(cfg).unwrap(); let ping = PingBuilder::new("store1").build(); diff --git a/glean-core/tests/ping.rs b/glean-core/tests/ping.rs index 884f25d158..7a7015441b 100644 --- a/glean-core/tests/ping.rs +++ b/glean-core/tests/ping.rs @@ -389,6 +389,7 @@ fn clearing_storage_by_prefix_doesnt_clear_unrelated_delayed_ping_io() { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }; let mut glean = glean_core::Glean::new(cfg).unwrap(); diff --git a/glean-core/tests/ping_maker.rs b/glean-core/tests/ping_maker.rs index 0bc84b3aed..75d1be3204 100644 --- a/glean-core/tests/ping_maker.rs +++ b/glean-core/tests/ping_maker.rs @@ -102,6 +102,7 @@ fn test_metrics_must_report_experimentation_id() { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }) .unwrap(); let ping_maker = PingMaker::new(); @@ -163,6 +164,7 @@ fn experimentation_id_is_removed_if_send_if_empty_is_false() { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, }) .unwrap(); let ping_maker = PingMaker::new(); diff --git a/glean-core/tests/session.rs b/glean-core/tests/session.rs index 4c51a6da49..c857da2234 100644 --- a/glean-core/tests/session.rs +++ b/glean-core/tests/session.rs @@ -47,6 +47,7 @@ fn session_cfg( session_sample_rate: sample_rate, session_inactivity_timeout_ms: timeout_ms, events_ping_acceleration_factor: None, + enable_store_submitted_pings: false, } } diff --git a/glean-core/tests/sqlite.rs b/glean-core/tests/sqlite.rs index e65a5751d3..086011374b 100644 --- a/glean-core/tests/sqlite.rs +++ b/glean-core/tests/sqlite.rs @@ -3,9 +3,9 @@ // file, You can obtain one at https://mozilla.org/MPL/2.0/. mod common; -use std::fs; - use crate::common::*; +use chrono::Utc; +use std::fs; use glean_core::metrics::*; use glean_core::CommonMetricData; @@ -205,6 +205,7 @@ mod unix { session_sample_rate: 1.0, session_inactivity_timeout_ms: 1_800_000, events_ping_acceleration_factor: None, + enable_store_submitted_pings: true, }; let glean = Glean::new(cfg); assert!(glean.is_err()); @@ -253,13 +254,14 @@ fn database_externally_locked() { session_mode: SessionMode::Auto, session_sample_rate: 1.0, events_ping_acceleration_factor: None, + enable_store_submitted_pings: true, }; let glean = Glean::new(cfg); assert!(glean.is_err()); } #[test] -fn schema_v2_is_applied() { +fn latest_schema_is_applied() { let (first_client_id, temp) = { let (glean, temp) = new_glean(None); let client_id = clientid_metric().get_value(&glean, None).unwrap(); @@ -286,10 +288,164 @@ fn schema_v2_is_applied() { let cur_user_version: u32 = conn .query_one("PRAGMA user_version", [], |row| row.get(0)) .unwrap(); - assert_eq!(cur_user_version, 2); + assert_eq!(cur_user_version, 3); let migration_state: String = conn .query_one("SELECT state FROM migration", [], |row| row.get(0)) .unwrap(); assert_eq!(migration_state, "done"); } + +#[test] +fn test_storing_and_fetching_submitted_pings() { + let (glean, _temp) = new_glean(None); + + let utc_time_one = chrono::DateTime::parse_from_rfc3339("2026-08-05T12:30:00.50Z") + .unwrap() + .to_utc(); + let utc_time_two = chrono::DateTime::parse_from_rfc3339("2026-08-05T12:30:00.51Z") + .unwrap() + .to_utc(); + let utc_time_three = chrono::DateTime::parse_from_rfc3339("2026-08-05T12:30:00.52Z") + .unwrap() + .to_utc(); + + // First ping, no upload date + glean + .storage() + .store_submitted_ping( + "id", + "ping", + utc_time_one, + None, + false, + serde_json::json!({ "test": "a value" }), + ) + .unwrap(); + + // Second ping, no upload date + glean + .storage() + .store_submitted_ping( + "id-one", + "ping-two", + utc_time_two, + None, + false, + serde_json::json!({ "test": "a value" }), + ) + .unwrap(); + + // Second ping again, with upload date .01s after submitted date + glean + .storage() + .store_submitted_ping( + "id-one", + "ping-two", + utc_time_two, + Some(utc_time_two), + false, + serde_json::json!({ "test": "a value" }), + ) + .unwrap(); + + // Third ping, upload failed + glean + .storage() + .store_submitted_ping( + "id-two", + "ping-three", + utc_time_three, + None, + true, + serde_json::json!({ "test": "a value" }), + ) + .unwrap(); + + let all_pings = glean.storage().get_all_submitted_pings(); + assert_eq!(all_pings.len(), 3); + assert_eq!(all_pings.last().unwrap().document_id, "id".to_string()); + assert_eq!(all_pings.get(1).unwrap().document_id, "id-one".to_string()); + assert_eq!(all_pings.get(1).unwrap().submitted_date.0, utc_time_two); + assert_eq!( + all_pings.get(1).unwrap().uploaded_date.clone().unwrap().0, + utc_time_two + ); + assert_eq!( + all_pings.get(1).unwrap().payload().unwrap(), + serde_json::json!({ "test": "a value" }) + ); + assert!(all_pings.first().unwrap().upload_failed); + + let count = glean.storage().mark_ping_as_uploaded("id", utc_time_one); + assert_eq!(count, 1); + + let some_pings = glean.storage().get_submitted_pings_by_name("ping"); + assert_eq!(some_pings.len(), 1); + assert_eq!(some_pings.first().unwrap().document_id, "id".to_string()); + assert_eq!( + some_pings.first().unwrap().uploaded_date.clone().unwrap().0, + utc_time_one + ); +} + +#[test] +fn test_cleanup_of_submitted_pings() { + let (glean, _temp) = new_glean(None); + + let utc_time_more_than_30_days_ago = + chrono::DateTime::parse_from_rfc3339("2026-06-05T12:30:00.50Z") + .unwrap() + .to_utc(); + + // Submitted ping from more than 30 days ago + glean + .storage() + .store_submitted_ping( + "id-one", + "ping", + utc_time_more_than_30_days_ago, + None, + false, + serde_json::json!({ "test": "a value" }), + ) + .unwrap(); + + // Submitted ping from now + glean + .storage() + .store_submitted_ping( + "id-two", + "ping", + Utc::now(), + None, + false, + serde_json::json!({ "test": "a value" }), + ) + .unwrap(); + + // Both pings should have been stored + let all_pings = glean.storage().get_all_submitted_pings(); + assert_eq!(all_pings.len(), 2); + + // Run regular maintenance (happens on shutdown) + // This should only remove the ping from >30 days ago + glean + .storage() + .cleanup_submitted_pings(None) + .expect("Error running cleanup_submitted_pings"); + + let all_pings = glean.storage().get_all_submitted_pings(); + assert_eq!(all_pings.len(), 1); + assert_eq!(all_pings.first().unwrap().document_id, "id-two".to_string()); + + // Run `cleanup_submitted_pings` with now as the `before_time` + // This should clear out all pings + glean + .storage() + .cleanup_submitted_pings(Some(Utc::now())) + .expect("Error running cleanup_submitted_pings"); + + let all_pings = glean.storage().get_all_submitted_pings(); + assert_eq!(all_pings.len(), 0); +}