WIP for collecting send-tab telemetry.

components/fxa-client/Cargo.toml

@@ -26,6 +26,7 @@ rc_crypto = { path = "../support/rc_crypto", features = ["ece", "hawk"] }
 error-support = { path = "../support/error" }
 thiserror = "1.0"
 anyhow = "1.0"
+sync-guid = { path = "../support/guid", features = ["random"] }
 
 [dev-dependencies]
 viaduct-reqwest = { path = "../support/viaduct-reqwest" }

components/fxa-client/android/src/main/java/mozilla/appservices/fxaclient/FirefoxAccount.kt

@@ -632,6 +632,18 @@ class FirefoxAccount(handle: FxaHandle, persistCallback: PersistCallback?) : Aut
         }
     }
 
+    /**
+     * Gather any telemetry which has been collected internally and return
+     * the result as a JSON string.
+     *
+     * This does not make network requests, and can be used on the main thread.
+     */
+    fun gatherTelemetry(): String {
+        return rustCallWithLock { e ->
+            LibFxAFFI.INSTANCE.fxa_gather_telemetry(this.handle.get(), e)
+        }.getAndConsumeRustString()
+    }
+
     @Synchronized
     override fun close() {
         val handle = this.handle.getAndSet(0)

components/fxa-client/android/src/main/java/mozilla/appservices/fxaclient/rust/LibFxAFFI.kt

@@ -114,6 +114,8 @@ internal interface LibFxAFFI : Library {
 
     fun fxa_retry_migrate_from_session_token(fxa: FxaHandle, e: RustError.ByReference): Pointer?
 
+    fun fxa_gather_telemetry(fxa: FxaHandle, e: RustError.ByReference): Pointer?
+
     fun fxa_str_free(string: Pointer)
     fun fxa_bytebuffer_free(buffer: RustBuffer.ByValue)
     fun fxa_free(fxa: FxaHandle, err: RustError.ByReference)

components/fxa-client/ffi/src/lib.rs

@@ -12,7 +12,7 @@ use ffi_support::{
     ConcurrentHandleMap, ExternError, FfiStr,
 };
 use fxa_client::{
-    device::{Capability as DeviceCapability, PushSubscription},
+    device::{Capability as DeviceCapability, CommandFetchReason, PushSubscription},
     ffi::{from_protobuf_ptr, AuthorizationParameters, MetricsParams},
     migrator::MigrationState,
     msg_types, FirefoxAccount,
@@ -506,10 +506,11 @@ pub extern "C" fn fxa_handle_session_token_change(
 pub extern "C" fn fxa_poll_device_commands(handle: u64, error: &mut ExternError) -> ByteBuffer {
     log::debug!("fxa_poll_device_commands");
     ACCOUNTS.call_with_result_mut(error, handle, |fxa| {
-        fxa.poll_device_commands().map(|cmds| {
-            let commands = cmds.into_iter().map(|e| e.into()).collect();
-            fxa_client::msg_types::IncomingDeviceCommands { commands }
-        })
+        fxa.poll_device_commands(CommandFetchReason::Poll)
+            .map(|cmds| {
+                let commands = cmds.into_iter().map(|e| e.into()).collect();
+                fxa_client::msg_types::IncomingDeviceCommands { commands }
+            })
     })
 }
 
@@ -615,3 +616,10 @@ pub extern "C" fn fxa_get_ecosystem_anon_id(handle: u64, error: &mut ExternError
 define_handle_map_deleter!(ACCOUNTS, fxa_free);
 define_string_destructor!(fxa_str_free);
 define_bytebuffer_destructor!(fxa_bytebuffer_free);
+
+/// Gather telemetry collected by FxA.
+#[no_mangle]
+pub extern "C" fn fxa_gather_telemetry(handle: u64, error: &mut ExternError) -> *mut c_char {
+    log::debug!("fxa_gather_telemetry");
+    ACCOUNTS.call_with_result_mut(error, handle, |fxa| fxa.gather_telemetry())
+}

components/fxa-client/ios/FxAClient/RustFxAFFI.h

@@ -165,6 +165,9 @@ char *_Nullable fxa_get_manage_devices_url(FirefoxAccountHandle handle,
                                            const char *_Nonnull entrypoint,
                                            FxAError *_Nonnull out);
 
+char *fxa_gather_telemetry(FirefoxAccountHandle handle,
+                           FxAError *_Nonnull out);
+
 void fxa_str_free(char *_Nullable ptr);
 void fxa_free(FirefoxAccountHandle h, FxAError *_Nonnull out);
 void fxa_bytebuffer_free(FxARustBuffer buffer);

components/fxa-client/src/commands/send_tab.rs

@@ -13,7 +13,7 @@
 /// uses the obtained public key to encrypt the `SendTabPayload` it created that
 /// contains the tab to send and finally forms the `EncryptedSendTabPayload` that is
 /// then sent to the target device.
-use crate::{device::Device, error::*, scoped_keys::ScopedKey, scopes};
+use crate::{device::Device, error::*, scoped_keys::ScopedKey, scopes, telemetry};
 use rc_crypto::ece::{self, Aes128GcmEceWebPush, EcKeyComponents, WebPushParams};
 use rc_crypto::ece_crypto::{RcCryptoLocalKeyPair, RcCryptoRemotePublicKey};
 use serde_derive::*;
@@ -40,16 +40,26 @@ impl EncryptedSendTabPayload {
 #[derive(Debug, Serialize, Deserialize)]
 pub struct SendTabPayload {
     pub entries: Vec<TabHistoryEntry>,
+    #[serde(rename = "flowID", default)]
+    pub flow_id: String,
+    #[serde(rename = "streamID", default)]
+    pub stream_id: String,
 }
 
 impl SendTabPayload {
-    pub fn single_tab(title: &str, url: &str) -> Self {
-        SendTabPayload {
-            entries: vec![TabHistoryEntry {
-                title: title.to_string(),
-                url: url.to_string(),
-            }],
-        }
+    pub fn single_tab(title: &str, url: &str) -> (Self, telemetry::SentCommand) {
+        let sent_telemetry: telemetry::SentCommand = Default::default();
+        (
+            SendTabPayload {
+                entries: vec![TabHistoryEntry {
+                    title: title.to_string(),
+                    url: url.to_string(),
+                }],
+                flow_id: sent_telemetry.flow_id.clone(),
+                stream_id: sent_telemetry.stream_id.clone(),
+            },
+            sent_telemetry,
+        )
     }
     fn encrypt(&self, keys: PublicSendTabKeys) -> Result<EncryptedSendTabPayload> {
         rc_crypto::ensure_initialized();
@@ -218,3 +228,29 @@ fn extract_oldsync_key_components(oldsync_key: &ScopedKey) -> Result<(Vec<u8>, V
     let ksync = oldsync_key.key_bytes()?;
     Ok((ksync, kxcs))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_minimal_parse_payload() {
+        let minimal = r#"{ "entries": []}"#;
+        let payload: SendTabPayload = serde_json::from_str(minimal).expect("should work");
+        assert_eq!(payload.flow_id, "".to_string());
+    }
+
+    #[test]
+    fn test_payload() {
+        let (payload, telem) = SendTabPayload::single_tab("title", "http://example.com");
+        let json = serde_json::to_string(&payload).expect("should work");
+        assert_eq!(telem.flow_id.len(), 12);
+        assert_eq!(telem.stream_id.len(), 12);
+        assert_ne!(telem.flow_id, telem.stream_id);
+        let p2: SendTabPayload = serde_json::from_str(&json).expect("should work");
+        // no 'PartialEq' derived so check each field individually...
+        assert_eq!(payload.entries[0].url, "http://example.com".to_string());
+        assert_eq!(payload.flow_id, p2.flow_id);
+        assert_eq!(payload.stream_id, p2.stream_id);
+    }
+}

components/fxa-client/src/device.rs

@@ -9,17 +9,25 @@ use crate::{
     commands,
     error::*,
     http_client::{
-        CommandData, DeviceUpdateRequest, DeviceUpdateRequestBuilder, PendingCommand,
-        UpdateDeviceResponse,
+        DeviceUpdateRequest, DeviceUpdateRequestBuilder, PendingCommand, UpdateDeviceResponse,
     },
-    util, CachedResponse, FirefoxAccount, IncomingDeviceCommand,
+    telemetry, util, CachedResponse, FirefoxAccount, IncomingDeviceCommand,
 };
 use serde_derive::*;
 use std::collections::{HashMap, HashSet};
 
 // An devices response is considered fresh for `DEVICES_FRESHNESS_THRESHOLD` ms.
 const DEVICES_FRESHNESS_THRESHOLD: u64 = 60_000; // 1 minute
 
+/// The reason we are fetching commands.
+#[derive(Clone, Copy)]
+pub enum CommandFetchReason {
+    /// We are polling in-case we've missed some.
+    Poll,
+    /// We got a push notification with the index of the message.
+    Push(u64),
+}
+
 impl FirefoxAccount {
     /// Fetches the list of devices from the current account including
     /// the current one.
@@ -162,19 +170,33 @@ impl FirefoxAccount {
     /// Poll and parse any pending available command for our device.
     /// This should be called semi-regularly as the main method of
     /// commands delivery (push) can sometimes be unreliable on mobile devices.
+    /// Typically called even when a push notification is received, so that
+    /// any prior messages for which a push didn't arrive are still handled.
     ///
     /// **💾 This method alters the persisted account state.**
-    pub fn poll_device_commands(&mut self) -> Result<Vec<IncomingDeviceCommand>> {
+    pub fn poll_device_commands(
+        &mut self,
+        reason: CommandFetchReason,
+    ) -> Result<Vec<IncomingDeviceCommand>> {
         let last_command_index = self.state.last_handled_command.unwrap_or(0);
         // We increment last_command_index by 1 because the server response includes the current index.
-        self.fetch_and_parse_commands(last_command_index + 1, None)
+        self.fetch_and_parse_commands(last_command_index + 1, None, reason)
     }
 
     /// Retrieve and parse a specific command designated by its index.
     ///
     /// **💾 This method alters the persisted account state.**
-    pub fn fetch_device_command(&mut self, index: u64) -> Result<IncomingDeviceCommand> {
-        let mut device_commands = self.fetch_and_parse_commands(index, Some(1))?;
+    ///
+    /// Note that this should not be used if possible, as it does not correctly
+    /// handle missed messages. It's currently used only on iOS due to platform
+    /// restrictions (but we should still try and work out how to correctly
+    /// handle missed messages within those restrictions)
+    /// (What's wrong: if we get a push for tab-1 and a push for tab-3, and
+    /// between them I've never explicitly polled, I'll miss tab-2, even if I
+    /// try polling now)
+    pub fn ios_fetch_device_command(&mut self, index: u64) -> Result<IncomingDeviceCommand> {
+        let mut device_commands =
+            self.fetch_and_parse_commands(index, Some(1), CommandFetchReason::Push(index))?;
         let device_command = device_commands
             .pop()
             .ok_or_else(|| ErrorKind::IllegalState("Index fetch came out empty."))?;
@@ -188,6 +210,7 @@ impl FirefoxAccount {
         &mut self,
         index: u64,
         limit: Option<u64>,
+        reason: CommandFetchReason,
     ) -> Result<Vec<IncomingDeviceCommand>> {
         let refresh_token = self.get_refresh_token()?;
         let pending_commands =
@@ -197,19 +220,20 @@ impl FirefoxAccount {
             return Ok(Vec::new());
         }
         log::info!("Handling {} messages", pending_commands.messages.len());
-        let device_commands = self.parse_commands_messages(pending_commands.messages)?;
+        let device_commands = self.parse_commands_messages(pending_commands.messages, reason)?;
         self.state.last_handled_command = Some(pending_commands.index);
         Ok(device_commands)
     }
 
     fn parse_commands_messages(
         &mut self,
         messages: Vec<PendingCommand>,
+        reason: CommandFetchReason,
     ) -> Result<Vec<IncomingDeviceCommand>> {
         let devices = self.get_devices(false)?;
         let parsed_commands = messages
             .into_iter()
-            .filter_map(|msg| match self.parse_command(msg.data, &devices) {
+            .filter_map(|msg| match self.parse_command(msg, &devices, reason) {
                 Ok(device_command) => Some(device_command),
                 Err(e) => {
                     log::error!("Error while processing command: {}", e);
@@ -222,15 +246,24 @@ impl FirefoxAccount {
 
     fn parse_command(
         &mut self,
-        command_data: CommandData,
+        command: PendingCommand,
         devices: &[Device],
+        reason: CommandFetchReason,
     ) -> Result<IncomingDeviceCommand> {
+        let telem_reason = match reason {
+            CommandFetchReason::Poll => telemetry::ReceivedReason::Poll,
+            CommandFetchReason::Push(index) if command.index < index => {
+                telemetry::ReceivedReason::PushMissed
+            }
+            _ => telemetry::ReceivedReason::Push,
+        };
+        let command_data = command.data;
         let sender = command_data
             .sender
             .and_then(|s| devices.iter().find(|i| i.id == s).cloned());
         match command_data.command.as_str() {
             commands::send_tab::COMMAND_NAME => {
-                self.handle_send_tab_command(sender, command_data.payload)
+                self.handle_send_tab_command(sender, command_data.payload, telem_reason)
             }
             _ => Err(ErrorKind::UnknownCommand(command_data.command).into()),
         }

components/fxa-client/src/lib.rs

@@ -18,9 +18,11 @@ pub use crate::{
     oauth::IntrospectInfo,
     oauth::{AccessTokenInfo, RefreshToken},
     profile::Profile,
+    telemetry::FxaTelemetry,
 };
 use serde_derive::*;
 use std::{
+    cell::RefCell,
     collections::{HashMap, HashSet},
     sync::Arc,
 };
@@ -66,6 +68,9 @@ pub struct FirefoxAccount {
     flow_store: HashMap<String, OAuthFlow>,
     attached_clients_cache: Option<CachedResponse<Vec<http_client::GetAttachedClientResponse>>>,
     devices_cache: Option<CachedResponse<Vec<http_client::GetDeviceResponse>>>,
+    // 'telemetry' is only currently used by `&mut self` functions, but that's
+    // not something we want to insist on going forward, so RefCell<> it.
+    telemetry: RefCell<FxaTelemetry>,
 }
 
 impl FirefoxAccount {
@@ -76,6 +81,7 @@ impl FirefoxAccount {
             flow_store: HashMap::new(),
             attached_clients_cache: None,
             devices_cache: None,
+            telemetry: RefCell::new(FxaTelemetry::new()),
         }
     }
 
@@ -152,6 +158,7 @@ impl FirefoxAccount {
         self.state = self.state.start_over();
         self.flow_store.clear();
         self.clear_devices_and_attached_clients_cache();
+        self.telemetry.replace(FxaTelemetry::new());
     }
 
     /// Get the Sync Token Server endpoint URL.

components/fxa-client/src/push.rs

@@ -2,6 +2,7 @@
  * License, v. 2.0. If a copy of the MPL was not distributed with this
  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
 
+use crate::device::CommandFetchReason;
 use crate::{error::*, AccountEvent, FirefoxAccount};
 use serde_derive::Deserialize;
 
@@ -13,6 +14,8 @@ impl FirefoxAccount {
     /// Since FxA sends one push notification per command received,
     /// we must only retrieve 1 command per push message,
     /// otherwise we risk receiving push messages for which the UI has already been shown.
+    /// However, note that this means iOS currently risks losing messages for
+    /// which a push notification doesn't arrive.
     ///
     /// **💾 This method alters the persisted account state.**
     pub fn handle_push_message(&mut self, payload: &str) -> Result<Vec<AccountEvent>> {
@@ -30,14 +33,15 @@ impl FirefoxAccount {
         match payload {
             PushPayload::CommandReceived(CommandReceivedPushPayload { index, .. }) => {
                 if cfg!(target_os = "ios") {
-                    self.fetch_device_command(index)
+                    self.ios_fetch_device_command(index)
                         .map(|cmd| vec![AccountEvent::IncomingDeviceCommand(Box::new(cmd))])
                 } else {
-                    self.poll_device_commands().map(|cmds| {
-                        cmds.into_iter()
-                            .map(|cmd| AccountEvent::IncomingDeviceCommand(Box::new(cmd)))
-                            .collect()
-                    })
+                    self.poll_device_commands(CommandFetchReason::Push(index))
+                        .map(|cmds| {
+                            cmds.into_iter()
+                                .map(|cmd| AccountEvent::IncomingDeviceCommand(Box::new(cmd)))
+                                .collect()
+                        })
                 }
             }
             PushPayload::ProfileUpdated => {