summaryrefslogtreecommitdiffstats
path: root/third_party/rust/sync15/src/clients
diff options
context:
space:
mode:
Diffstat (limited to 'third_party/rust/sync15/src/clients')
-rw-r--r--third_party/rust/sync15/src/clients/engine.rs706
-rw-r--r--third_party/rust/sync15/src/clients/mod.rs94
-rw-r--r--third_party/rust/sync15/src/clients/record.rs158
-rw-r--r--third_party/rust/sync15/src/clients/ser.rs125
4 files changed, 1083 insertions, 0 deletions
diff --git a/third_party/rust/sync15/src/clients/engine.rs b/third_party/rust/sync15/src/clients/engine.rs
new file mode 100644
index 0000000000..a9e77d109d
--- /dev/null
+++ b/third_party/rust/sync15/src/clients/engine.rs
@@ -0,0 +1,706 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * 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 std::collections::{HashMap, HashSet};
+
+use crate::{
+ bso_record::Payload,
+ changeset::{CollectionUpdate, IncomingChangeset, OutgoingChangeset},
+ client::Sync15StorageClient,
+ coll_state::CollState,
+ collection_keys::CollectionKeys,
+ key_bundle::KeyBundle,
+ request::{CollectionRequest, InfoConfiguration},
+ state::GlobalState,
+};
+use interrupt_support::Interruptee;
+use sync15_traits::client::ClientData;
+
+use super::{
+ record::{ClientRecord, CommandRecord},
+ ser::shrink_to_fit,
+ Command, CommandProcessor, CommandStatus, RemoteClient, CLIENTS_TTL,
+};
+use crate::error::Result;
+
+const COLLECTION_NAME: &str = "clients";
+
+/// The driver for the clients engine. Internal; split out from the `Engine`
+/// struct to make testing easier.
+struct Driver<'a> {
+ command_processor: &'a dyn CommandProcessor,
+ interruptee: &'a dyn Interruptee,
+ config: &'a InfoConfiguration,
+ recent_clients: HashMap<String, RemoteClient>,
+}
+
+impl<'a> Driver<'a> {
+ fn new(
+ command_processor: &'a dyn CommandProcessor,
+ interruptee: &'a dyn Interruptee,
+ config: &'a InfoConfiguration,
+ ) -> Driver<'a> {
+ Driver {
+ command_processor,
+ interruptee,
+ config,
+ recent_clients: HashMap::new(),
+ }
+ }
+
+ fn note_recent_client(&mut self, client: &ClientRecord) {
+ self.recent_clients.insert(client.id.clone(), client.into());
+ }
+
+ fn sync(
+ &mut self,
+ inbound: IncomingChangeset,
+ should_refresh_client: bool,
+ ) -> Result<OutgoingChangeset> {
+ let mut outgoing = OutgoingChangeset::new(COLLECTION_NAME, inbound.timestamp);
+ outgoing.timestamp = inbound.timestamp;
+
+ self.interruptee.err_if_interrupted()?;
+ let outgoing_commands = self.command_processor.fetch_outgoing_commands()?;
+
+ let mut has_own_client_record = false;
+
+ for (payload, _) in inbound.changes {
+ self.interruptee.err_if_interrupted()?;
+
+ // Unpack the client record. We should never have tombstones in the
+ // clients collection, so we don't check for `is_tombstone`.
+ // https://github.com/mozilla/application-services/issues/1801
+ // tracks deleting these from the server.
+ let client: ClientRecord = payload.into_record()?;
+
+ if client.id == self.command_processor.settings().fxa_device_id {
+ log::debug!("Found my record on the server");
+ // If we see our own client record, apply any incoming commands,
+ // remove them from the list, and reupload the record. Any
+ // commands that we don't understand also go back in the list.
+ // https://github.com/mozilla/application-services/issues/1800
+ // tracks if that's the right thing to do.
+ has_own_client_record = true;
+ let mut current_client_record = self.current_client_record();
+ for c in &client.commands {
+ let status = match c.as_command() {
+ Some(command) => self.command_processor.apply_incoming_command(command)?,
+ None => CommandStatus::Unsupported,
+ };
+ match status {
+ CommandStatus::Applied => {}
+ CommandStatus::Ignored => {
+ log::debug!("Ignored command {:?}", c);
+ }
+ CommandStatus::Unsupported => {
+ log::warn!("Don't know how to apply command {:?}", c);
+ current_client_record.commands.push(c.clone());
+ }
+ }
+ }
+
+ // The clients collection has a hard limit on the payload size,
+ // after which the server starts rejecting our records. Large
+ // command lists can cause us to exceed this, so we truncate
+ // the list.
+ shrink_to_fit(
+ &mut current_client_record.commands,
+ self.memcache_max_record_payload_size(),
+ )?;
+
+ // Add the new client record to our map of recently synced
+ // clients, so that downstream consumers like synced tabs can
+ // access them.
+ self.note_recent_client(&current_client_record);
+
+ // We periodically upload our own client record, even if it
+ // doesn't change, to keep it fresh.
+ // (but this part sucks - if the ttl on the server happens to be
+ // different (as some other client did something strange) we
+ // still want the records to compare equal - but the ttl hack
+ // doesn't allow that.)
+ let mut client_compare = client.clone();
+ client_compare.ttl = current_client_record.ttl;
+ if should_refresh_client || client_compare != current_client_record {
+ log::debug!("Will update our client record on the server");
+ outgoing
+ .changes
+ .push(Payload::from_record(current_client_record)?);
+ }
+ } else {
+ // Add the other client to our map of recently synced clients.
+ self.note_recent_client(&client);
+
+ // Bail if we don't have any outgoing commands to write into
+ // the other client's record.
+ if outgoing_commands.is_empty() {
+ continue;
+ }
+
+ // Determine if we have new commands, that aren't already in the
+ // client's command list.
+ let current_commands: HashSet<Command> = client
+ .commands
+ .iter()
+ .filter_map(|c| c.as_command())
+ .collect();
+ let mut new_outgoing_commands = outgoing_commands
+ .difference(&current_commands)
+ .cloned()
+ .collect::<Vec<_>>();
+ // Sort, to ensure deterministic ordering for tests.
+ new_outgoing_commands.sort();
+ let mut new_client = client.clone();
+ new_client
+ .commands
+ .extend(new_outgoing_commands.into_iter().map(CommandRecord::from));
+ if new_client.commands.len() == client.commands.len() {
+ continue;
+ }
+
+ // Hooray, we added new commands! Make sure the record still
+ // fits in the maximum record size, or the server will reject
+ // our upload.
+ shrink_to_fit(
+ &mut new_client.commands,
+ self.memcache_max_record_payload_size(),
+ )?;
+
+ // We want to ensure the TTL for all records we write, which
+ // may not be true for incoming ones - so make sure it is.
+ new_client.ttl = CLIENTS_TTL;
+ outgoing.changes.push(Payload::from_record(new_client)?);
+ }
+ }
+
+ // Upload a record for our own client, if we didn't replace it already.
+ if !has_own_client_record {
+ let current_client_record = self.current_client_record();
+ self.note_recent_client(&current_client_record);
+ outgoing
+ .changes
+ .push(Payload::from_record(current_client_record)?);
+ }
+
+ Ok(outgoing)
+ }
+
+ /// Builds a fresh client record for this device.
+ fn current_client_record(&self) -> ClientRecord {
+ let settings = self.command_processor.settings();
+ ClientRecord {
+ id: settings.fxa_device_id.clone(),
+ name: settings.device_name.clone(),
+ typ: Some(settings.device_type.as_str().into()),
+ commands: Vec::new(),
+ fxa_device_id: Some(settings.fxa_device_id.clone()),
+ version: None,
+ protocols: vec!["1.5".into()],
+ form_factor: None,
+ os: None,
+ app_package: None,
+ application: None,
+ device: None,
+ ttl: CLIENTS_TTL,
+ }
+ }
+
+ fn max_record_payload_size(&self) -> usize {
+ let payload_max = self.config.max_record_payload_bytes;
+ if payload_max <= self.config.max_post_bytes {
+ self.config.max_post_bytes.saturating_sub(4096)
+ } else {
+ payload_max
+ }
+ }
+
+ /// Collections stored in memcached ("tabs", "clients" or "meta") have a
+ /// different max size than ones stored in the normal storage server db.
+ /// In practice, the real limit here is 1M (bug 1300451 comment 40), but
+ /// there's overhead involved that is hard to calculate on the client, so we
+ /// use 512k to be safe (at the recommendation of the server team). Note
+ /// that if the server reports a lower limit (via info/configuration), we
+ /// respect that limit instead. See also bug 1403052.
+ fn memcache_max_record_payload_size(&self) -> usize {
+ self.max_record_payload_size().min(512 * 1024)
+ }
+}
+
+pub struct Engine<'a> {
+ pub command_processor: &'a dyn CommandProcessor,
+ pub interruptee: &'a dyn Interruptee,
+ pub recent_clients: HashMap<String, RemoteClient>,
+}
+
+impl<'a> Engine<'a> {
+ /// Creates a new clients engine that delegates to the given command
+ /// processor to apply incoming commands.
+ pub fn new<'b>(
+ command_processor: &'b dyn CommandProcessor,
+ interruptee: &'b dyn Interruptee,
+ ) -> Engine<'b> {
+ Engine {
+ command_processor,
+ interruptee,
+ recent_clients: HashMap::new(),
+ }
+ }
+
+ /// Syncs the clients collection. This works a little differently than
+ /// other collections:
+ ///
+ /// 1. It can't be disabled or declined.
+ /// 2. The sync ID and last sync time aren't meaningful, since we always
+ /// fetch all client records on every sync. As such, the
+ /// `LocalCollStateMachine` that we use for other engines doesn't
+ /// apply to it.
+ /// 3. It doesn't persist state directly, but relies on the sync manager
+ /// to persist device settings, and process commands.
+ /// 4. Failing to sync the clients collection is fatal, and aborts the
+ /// sync.
+ ///
+ /// For these reasons, we implement this engine directly in the `sync15`
+ /// crate, and provide a specialized `sync` method instead of implementing
+ /// `sync15::Store`.
+ pub fn sync(
+ &mut self,
+ storage_client: &Sync15StorageClient,
+ global_state: &GlobalState,
+ root_sync_key: &KeyBundle,
+ should_refresh_client: bool,
+ ) -> Result<()> {
+ log::info!("Syncing collection clients");
+
+ let coll_keys =
+ CollectionKeys::from_encrypted_bso(global_state.keys.clone(), &root_sync_key)?;
+ let mut coll_state = CollState {
+ config: global_state.config.clone(),
+ last_modified: global_state
+ .collections
+ .get(COLLECTION_NAME)
+ .cloned()
+ .unwrap_or_default(),
+ key: coll_keys.key_for_collection(COLLECTION_NAME).clone(),
+ };
+
+ let inbound = self.fetch_incoming(&storage_client, &mut coll_state)?;
+
+ let mut driver = Driver::new(
+ self.command_processor,
+ self.interruptee,
+ &global_state.config,
+ );
+
+ let outgoing = driver.sync(inbound, should_refresh_client)?;
+ self.recent_clients = driver.recent_clients;
+
+ coll_state.last_modified = outgoing.timestamp;
+
+ self.interruptee.err_if_interrupted()?;
+ let upload_info =
+ CollectionUpdate::new_from_changeset(&storage_client, &coll_state, outgoing, true)?
+ .upload()?;
+
+ log::info!(
+ "Upload success ({} records success, {} records failed)",
+ upload_info.successful_ids.len(),
+ upload_info.failed_ids.len()
+ );
+
+ log::info!("Finished syncing clients");
+ Ok(())
+ }
+
+ fn fetch_incoming(
+ &self,
+ storage_client: &Sync15StorageClient,
+ coll_state: &mut CollState,
+ ) -> Result<IncomingChangeset> {
+ // Note that, unlike other stores, we always fetch the full collection
+ // on every sync, so `inbound` will return all clients, not just the
+ // ones that changed since the last sync.
+ let coll_request = CollectionRequest::new(COLLECTION_NAME).full();
+
+ self.interruptee.err_if_interrupted()?;
+ let inbound = crate::changeset::fetch_incoming(&storage_client, coll_state, &coll_request)?;
+
+ Ok(inbound)
+ }
+
+ pub fn local_client_id(&self) -> String {
+ // Bit dirty but it's the easiest way to reach to our own
+ // device ID without refactoring the whole sync manager crate.
+ self.command_processor.settings().fxa_device_id.clone()
+ }
+
+ pub fn get_client_data(&self) -> ClientData {
+ ClientData {
+ local_client_id: self.local_client_id(),
+ recent_clients: self.recent_clients.clone(),
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::clients::{CommandStatus, DeviceType, Settings};
+ use crate::util::ServerTimestamp;
+ use anyhow::Result;
+ use interrupt_support::NeverInterrupts;
+ use serde_json::{json, Value};
+
+ use super::*;
+
+ struct TestProcessor {
+ settings: Settings,
+ outgoing_commands: HashSet<Command>,
+ }
+
+ impl CommandProcessor for TestProcessor {
+ fn settings(&self) -> &Settings {
+ &self.settings
+ }
+
+ fn apply_incoming_command(&self, command: Command) -> Result<CommandStatus> {
+ Ok(if let Command::Reset(name) = command {
+ if name == "forms" {
+ CommandStatus::Unsupported
+ } else {
+ CommandStatus::Applied
+ }
+ } else {
+ CommandStatus::Ignored
+ })
+ }
+
+ fn fetch_outgoing_commands(&self) -> Result<HashSet<Command>> {
+ Ok(self.outgoing_commands.clone())
+ }
+ }
+
+ fn inbound_from_clients(clients: Value) -> IncomingChangeset {
+ if let Value::Array(clients) = clients {
+ let changes = clients
+ .into_iter()
+ .map(|c| (Payload::from_json(c).unwrap(), ServerTimestamp(0)))
+ .collect();
+ IncomingChangeset {
+ changes,
+ timestamp: ServerTimestamp(0),
+ collection: COLLECTION_NAME.into(),
+ }
+ } else {
+ unreachable!("`clients` must be an array of client records")
+ }
+ }
+
+ #[test]
+ fn test_clients_sync() {
+ let processor = TestProcessor {
+ settings: Settings {
+ fxa_device_id: "deviceAAAAAA".into(),
+ device_name: "Laptop".into(),
+ device_type: DeviceType::Desktop,
+ },
+ outgoing_commands: [
+ Command::Wipe("bookmarks".into()),
+ Command::Reset("history".into()),
+ ]
+ .iter()
+ .cloned()
+ .collect(),
+ };
+
+ let config = InfoConfiguration::default();
+
+ let mut driver = Driver::new(&processor, &NeverInterrupts, &config);
+
+ let inbound = inbound_from_clients(json!([{
+ "id": "deviceBBBBBB",
+ "name": "iPhone",
+ "type": "mobile",
+ "commands": [{
+ "command": "resetEngine",
+ "args": ["history"],
+ }],
+ "fxaDeviceId": "iPhooooooone",
+ "protocols": ["1.5"],
+ "device": "iPhone",
+ }, {
+ "id": "deviceCCCCCC",
+ "name": "Fenix",
+ "type": "mobile",
+ "commands": [],
+ "fxaDeviceId": "deviceCCCCCC",
+ }, {
+ "id": "deviceAAAAAA",
+ "name": "Laptop with a different name",
+ "type": "desktop",
+ "commands": [{
+ "command": "wipeEngine",
+ "args": ["logins"]
+ }, {
+ "command": "displayURI",
+ "args": ["http://example.com", "Fennec", "Example page"],
+ "flowID": "flooooooooow",
+ }, {
+ "command": "resetEngine",
+ "args": ["forms"],
+ }, {
+ "command": "logout",
+ "args": [],
+ }],
+ "fxaDeviceId": "deviceAAAAAA",
+ }]));
+
+ // Passing false for `should_refresh_client` - it should be ignored
+ // because we've changed the commands.
+ let mut outgoing = driver.sync(inbound, false).expect("Should sync clients");
+ outgoing.changes.sort_by(|a, b| a.id.cmp(&b.id));
+
+ // Make sure the list of recently synced remote clients is correct.
+ let expected_ids = &["deviceAAAAAA", "deviceBBBBBB", "deviceCCCCCC"];
+ let mut actual_ids = driver.recent_clients.keys().collect::<Vec<&String>>();
+ actual_ids.sort();
+ assert_eq!(actual_ids, expected_ids);
+
+ let expected_remote_clients = &[
+ RemoteClient {
+ fxa_device_id: Some("deviceAAAAAA".to_string()),
+ device_name: "Laptop".into(),
+ device_type: Some(DeviceType::Desktop),
+ },
+ RemoteClient {
+ fxa_device_id: Some("iPhooooooone".to_string()),
+ device_name: "iPhone".into(),
+ device_type: Some(DeviceType::Mobile),
+ },
+ RemoteClient {
+ fxa_device_id: Some("deviceCCCCCC".to_string()),
+ device_name: "Fenix".into(),
+ device_type: Some(DeviceType::Mobile),
+ },
+ ];
+ let actual_remote_clients = expected_ids
+ .iter()
+ .filter_map(|&id| driver.recent_clients.get(id))
+ .cloned()
+ .collect::<Vec<RemoteClient>>();
+ assert_eq!(actual_remote_clients, expected_remote_clients);
+
+ let expected = json!([{
+ "id": "deviceAAAAAA",
+ "name": "Laptop",
+ "type": "desktop",
+ "commands": [{
+ "command": "displayURI",
+ "args": ["http://example.com", "Fennec", "Example page"],
+ "flowID": "flooooooooow",
+ }, {
+ "command": "resetEngine",
+ "args": ["forms"],
+ }, {
+ "command": "logout",
+ "args": [],
+ }],
+ "fxaDeviceId": "deviceAAAAAA",
+ "protocols": ["1.5"],
+ "ttl": CLIENTS_TTL,
+ }, {
+ "id": "deviceBBBBBB",
+ "name": "iPhone",
+ "type": "mobile",
+ "commands": [{
+ "command": "resetEngine",
+ "args": ["history"],
+ }, {
+ "command": "wipeEngine",
+ "args": ["bookmarks"],
+ }],
+ "fxaDeviceId": "iPhooooooone",
+ "protocols": ["1.5"],
+ "device": "iPhone",
+ "ttl": CLIENTS_TTL,
+ }, {
+ "id": "deviceCCCCCC",
+ "name": "Fenix",
+ "type": "mobile",
+ "commands": [{
+ "command": "wipeEngine",
+ "args": ["bookmarks"],
+ }, {
+ "command": "resetEngine",
+ "args": ["history"],
+ }],
+ "fxaDeviceId": "deviceCCCCCC",
+ "ttl": CLIENTS_TTL,
+ }]);
+ if let Value::Array(expected) = expected {
+ for (i, record) in expected.into_iter().enumerate() {
+ assert_eq!(outgoing.changes[i], Payload::from_json(record).unwrap());
+ }
+ } else {
+ unreachable!("`expected_clients` must be an array of client records")
+ }
+ }
+
+ #[test]
+ fn test_clients_sync_explicit_refresh() {
+ let processor = TestProcessor {
+ settings: Settings {
+ fxa_device_id: "deviceAAAAAA".into(),
+ device_name: "Laptop".into(),
+ device_type: DeviceType::Desktop,
+ },
+ outgoing_commands: [].iter().cloned().collect(),
+ };
+
+ let config = InfoConfiguration::default();
+
+ let mut driver = Driver::new(&processor, &NeverInterrupts, &config);
+
+ let inbound = inbound_from_clients(json!([{
+ "id": "deviceBBBBBB",
+ "name": "iPhone",
+ "type": "mobile",
+ "commands": [{
+ "command": "resetEngine",
+ "args": ["history"],
+ }],
+ "fxaDeviceId": "iPhooooooone",
+ "protocols": ["1.5"],
+ "device": "iPhone",
+ "ttl": CLIENTS_TTL,
+ }, {
+ "id": "deviceAAAAAA",
+ "name": "Laptop",
+ "type": "desktop",
+ "commands": [],
+ "fxaDeviceId": "deviceAAAAAA",
+ "protocols": ["1.5"],
+ "ttl": CLIENTS_TTL,
+ }]));
+
+ let outgoing = driver
+ .sync(inbound.clone(), false)
+ .expect("Should sync clients");
+ // should be no outgoing changes.
+ assert_eq!(outgoing.changes.len(), 0);
+
+ // Make sure the list of recently synced remote clients is correct and
+ // still includes our record we didn't update.
+ let expected_ids = &["deviceAAAAAA", "deviceBBBBBB"];
+ let mut actual_ids = driver.recent_clients.keys().collect::<Vec<&String>>();
+ actual_ids.sort();
+ assert_eq!(actual_ids, expected_ids);
+
+ // Do it again - still no changes, but force a refresh.
+ let outgoing = driver.sync(inbound, true).expect("Should sync clients");
+ assert_eq!(outgoing.changes.len(), 1);
+
+ // Do it again - but this time with our own client record needing
+ // some change.
+ let inbound = inbound_from_clients(json!([{
+ "id": "deviceAAAAAA",
+ "name": "Laptop with New Name",
+ "type": "desktop",
+ "commands": [],
+ "fxaDeviceId": "deviceAAAAAA",
+ "protocols": ["1.5"],
+ }]));
+ let outgoing = driver.sync(inbound, false).expect("Should sync clients");
+ // should still be outgoing because the name changed.
+ assert_eq!(outgoing.changes.len(), 1);
+ }
+
+ #[test]
+ fn test_fresh_client_record() {
+ let processor = TestProcessor {
+ settings: Settings {
+ fxa_device_id: "deviceAAAAAA".into(),
+ device_name: "Laptop".into(),
+ device_type: DeviceType::Desktop,
+ },
+ outgoing_commands: HashSet::new(),
+ };
+
+ let config = InfoConfiguration::default();
+
+ let mut driver = Driver::new(&processor, &NeverInterrupts, &config);
+
+ let clients = json!([{
+ "id": "deviceBBBBBB",
+ "name": "iPhone",
+ "type": "mobile",
+ "commands": [{
+ "command": "resetEngine",
+ "args": ["history"],
+ }],
+ "fxaDeviceId": "iPhooooooone",
+ "protocols": ["1.5"],
+ "device": "iPhone",
+ }]);
+
+ let inbound = if let Value::Array(clients) = clients {
+ let changes = clients
+ .into_iter()
+ .map(|c| (Payload::from_json(c).unwrap(), ServerTimestamp(0)))
+ .collect();
+ IncomingChangeset {
+ changes,
+ timestamp: ServerTimestamp(0),
+ collection: COLLECTION_NAME.into(),
+ }
+ } else {
+ unreachable!("`clients` must be an array of client records")
+ };
+
+ // Passing false here for should_refresh_client, but it should be
+ // ignored as we don't have an existing record yet.
+ let mut outgoing = driver.sync(inbound, false).expect("Should sync clients");
+ outgoing.changes.sort_by(|a, b| a.id.cmp(&b.id));
+
+ // Make sure the list of recently synced remote clients is correct.
+ let expected_ids = &["deviceAAAAAA", "deviceBBBBBB"];
+ let mut actual_ids = driver.recent_clients.keys().collect::<Vec<&String>>();
+ actual_ids.sort();
+ assert_eq!(actual_ids, expected_ids);
+
+ let expected_remote_clients = &[
+ RemoteClient {
+ fxa_device_id: Some("deviceAAAAAA".to_string()),
+ device_name: "Laptop".into(),
+ device_type: Some(DeviceType::Desktop),
+ },
+ RemoteClient {
+ fxa_device_id: Some("iPhooooooone".to_string()),
+ device_name: "iPhone".into(),
+ device_type: Some(DeviceType::Mobile),
+ },
+ ];
+ let actual_remote_clients = expected_ids
+ .iter()
+ .filter_map(|&id| driver.recent_clients.get(id))
+ .cloned()
+ .collect::<Vec<RemoteClient>>();
+ assert_eq!(actual_remote_clients, expected_remote_clients);
+
+ let expected = json!([{
+ "id": "deviceAAAAAA",
+ "name": "Laptop",
+ "type": "desktop",
+ "fxaDeviceId": "deviceAAAAAA",
+ "protocols": ["1.5"],
+ "ttl": CLIENTS_TTL,
+ }]);
+ if let Value::Array(expected) = expected {
+ for (i, record) in expected.into_iter().enumerate() {
+ assert_eq!(outgoing.changes[i], Payload::from_json(record).unwrap());
+ }
+ } else {
+ unreachable!("`expected_clients` must be an array of client records")
+ }
+ }
+}
diff --git a/third_party/rust/sync15/src/clients/mod.rs b/third_party/rust/sync15/src/clients/mod.rs
new file mode 100644
index 0000000000..c1b7c9f32e
--- /dev/null
+++ b/third_party/rust/sync15/src/clients/mod.rs
@@ -0,0 +1,94 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * 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 std::collections::HashSet;
+
+mod engine;
+mod record;
+mod ser;
+
+use anyhow::Result;
+pub use engine::Engine;
+pub use sync15_traits::client::{ClientData, DeviceType, RemoteClient};
+
+// These are what desktop uses.
+const CLIENTS_TTL: u32 = 1_814_400; // 21 days
+pub(crate) const CLIENTS_TTL_REFRESH: u64 = 604_800; // 7 days
+
+/// A command processor applies incoming commands like wipes and resets for all
+/// stores, and returns commands to send to other clients. It also manages
+/// settings like the device name and type, which is stored in the special
+/// `clients` collection.
+///
+/// In practice, this trait only has one implementation, in the sync manager.
+/// It's split this way because the clients engine depends on internal `sync15`
+/// structures, and can't be implemented as a syncable store...but `sync15`
+/// doesn't know anything about multiple engines. This lets the sync manager
+/// provide its own implementation for handling wipe and reset commands for all
+/// the engines that it manages.
+pub trait CommandProcessor {
+ fn settings(&self) -> &Settings;
+
+ /// Fetches commands to send to other clients. An error return value means
+ /// commands couldn't be fetched, and halts the sync.
+ fn fetch_outgoing_commands(&self) -> Result<HashSet<Command>>;
+
+ /// Applies a command sent to this client from another client. This method
+ /// should return a `CommandStatus` indicating whether the command was
+ /// processed.
+ ///
+ /// An error return value means the sync manager encountered an error
+ /// applying the command, and halts the sync to prevent unexpected behavior
+ /// (for example, merging local and remote bookmarks, when we were told to
+ /// wipe our local bookmarks).
+ fn apply_incoming_command(&self, command: Command) -> Result<CommandStatus>;
+}
+
+/// Indicates if a command was applied successfully, ignored, or not supported.
+/// Applied and ignored commands are removed from our client record, and never
+/// retried. Unsupported commands are put back into our record, and retried on
+/// subsequent syncs. This is to handle clients adding support for new data
+/// types.
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub enum CommandStatus {
+ Applied,
+ Ignored,
+ Unsupported,
+}
+
+impl From<&record::ClientRecord> for RemoteClient {
+ fn from(record: &record::ClientRecord) -> RemoteClient {
+ RemoteClient {
+ fxa_device_id: record.fxa_device_id.clone(),
+ device_name: record.name.clone(),
+ device_type: record.typ.as_ref().and_then(DeviceType::try_from_str),
+ }
+ }
+}
+
+/// Information about this device to include in its client record. This should
+/// be persisted across syncs, as part of the sync manager state.
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub struct Settings {
+ /// The FxA device ID of this client, also used as this client's record ID
+ /// in the clients collection.
+ pub fxa_device_id: String,
+ /// The name of this client. This should match the client's name in the
+ /// FxA device manager.
+ pub device_name: String,
+ /// The type of this client: mobile, tablet, desktop, or other.
+ pub device_type: DeviceType,
+}
+
+#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub enum Command {
+ /// Erases all local data.
+ WipeAll,
+ /// Erases all local data for a specific engine.
+ Wipe(String),
+ /// Resets local sync state for all engines.
+ ResetAll,
+ /// Resets local sync state for a specific engine.
+ Reset(String),
+}
diff --git a/third_party/rust/sync15/src/clients/record.rs b/third_party/rust/sync15/src/clients/record.rs
new file mode 100644
index 0000000000..f263b5ac88
--- /dev/null
+++ b/third_party/rust/sync15/src/clients/record.rs
@@ -0,0 +1,158 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * 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 serde_derive::*;
+
+use super::Command;
+
+/// The serialized form of a client record.
+#[derive(Clone, Debug, Eq, Deserialize, Hash, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ClientRecord {
+ #[serde(rename = "id")]
+ pub id: String,
+
+ pub name: String,
+
+ #[serde(default, rename = "type")]
+ pub typ: Option<String>,
+
+ #[serde(default, skip_serializing_if = "Vec::is_empty")]
+ pub commands: Vec<CommandRecord>,
+
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub fxa_device_id: Option<String>,
+
+ /// `version`, `protocols`, `formfactor`, `os`, `appPackage`, `application`,
+ /// and `device` are unused and optional in all implementations (Desktop,
+ /// iOS, and Fennec), but we round-trip them.
+
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub version: Option<String>,
+
+ #[serde(default, skip_serializing_if = "Vec::is_empty")]
+ pub protocols: Vec<String>,
+
+ #[serde(
+ default,
+ rename = "formfactor",
+ skip_serializing_if = "Option::is_none"
+ )]
+ pub form_factor: Option<String>,
+
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub os: Option<String>,
+
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub app_package: Option<String>,
+
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub application: Option<String>,
+
+ /// The model of the device, like "iPhone" or "iPod touch" on iOS. Note
+ /// that this is _not_ the client ID (`id`) or the FxA device ID
+ /// (`fxa_device_id`).
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub device: Option<String>,
+
+ // This field is somewhat magic - it's moved to and from the
+ // BSO record, so is not expected to be on the unencrypted payload
+ // when incoming and are not put on the unencrypted payload when outgoing.
+ // There are hysterical raisens for this, which we should fix.
+ // https://github.com/mozilla/application-services/issues/2712
+ #[serde(default)]
+ pub ttl: u32,
+}
+
+/// The serialized form of a client command.
+#[derive(Clone, Debug, Eq, Deserialize, Hash, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CommandRecord {
+ /// The command name. This is a string, not an enum, because we want to
+ /// round-trip commands that we don't support yet.
+ #[serde(rename = "command")]
+ pub name: String,
+
+ /// Extra, command-specific arguments. Note that we must send an empty
+ /// array if the command expects no arguments.
+ #[serde(default)]
+ pub args: Vec<String>,
+
+ /// Some commands, like repair, send a "flow ID" that other cliennts can
+ /// record in their telemetry. We don't currently send commands with
+ /// flow IDs, but we round-trip them.
+ #[serde(default, rename = "flowID", skip_serializing_if = "Option::is_none")]
+ pub flow_id: Option<String>,
+}
+
+impl CommandRecord {
+ /// Converts a serialized command into one that we can apply. Returns `None`
+ /// if we don't support the command.
+ pub fn as_command(&self) -> Option<Command> {
+ match self.name.as_str() {
+ "wipeEngine" => self.args.get(0).map(|e| Command::Wipe(e.into())),
+ "wipeAll" => Some(Command::WipeAll),
+ "resetEngine" => self.args.get(0).map(|e| Command::Reset(e.into())),
+ "resetAll" => Some(Command::ResetAll),
+ _ => None,
+ }
+ }
+}
+
+impl From<Command> for CommandRecord {
+ fn from(command: Command) -> CommandRecord {
+ match command {
+ Command::Wipe(engine) => CommandRecord {
+ name: "wipeEngine".into(),
+ args: vec![engine],
+ flow_id: None,
+ },
+ Command::WipeAll => CommandRecord {
+ name: "wipeAll".into(),
+ args: Vec::new(),
+ flow_id: None,
+ },
+ Command::Reset(engine) => CommandRecord {
+ name: "resetEngine".into(),
+ args: vec![engine],
+ flow_id: None,
+ },
+ Command::ResetAll => CommandRecord {
+ name: "resetAll".into(),
+ args: Vec::new(),
+ flow_id: None,
+ },
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use sync15_traits::Payload;
+
+ #[test]
+ fn test_ttl() {
+ // The ttl hacks in place mean that magically the ttl field from the
+ // client record should make it down to a BSO.
+ let record = ClientRecord {
+ id: "id".into(),
+ name: "my device".into(),
+ typ: Some("type".into()),
+ commands: Vec::new(),
+ fxa_device_id: Some("12345".into()),
+ version: None,
+ protocols: vec!["1.5".into()],
+ form_factor: None,
+ os: None,
+ app_package: None,
+ application: None,
+ device: None,
+ ttl: 123,
+ };
+ let p = Payload::from_record(record).unwrap();
+ let bso = crate::CleartextBso::from_payload(p, "clients");
+ assert_eq!(bso.ttl, Some(123));
+ }
+}
diff --git a/third_party/rust/sync15/src/clients/ser.rs b/third_party/rust/sync15/src/clients/ser.rs
new file mode 100644
index 0000000000..0e8b85b0b8
--- /dev/null
+++ b/third_party/rust/sync15/src/clients/ser.rs
@@ -0,0 +1,125 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * 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::error::Result;
+use serde::Serialize;
+use std::io::{self, Write};
+
+/// A writer that counts the number of bytes it's asked to write, and discards
+/// the data. Used to calculate the serialized size of the commands list.
+#[derive(Clone, Copy, Default)]
+pub struct WriteCount(usize);
+
+impl WriteCount {
+ #[inline]
+ pub fn len(self) -> usize {
+ self.0
+ }
+}
+
+impl Write for WriteCount {
+ #[inline]
+ fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+ self.0 += buf.len();
+ Ok(buf.len())
+ }
+
+ #[inline]
+ fn flush(&mut self) -> io::Result<()> {
+ Ok(())
+ }
+}
+
+/// Returns the size of the given value, in bytes, when serialized to JSON.
+fn compute_serialized_size<T: Serialize>(value: &T) -> Result<usize> {
+ let mut w = WriteCount::default();
+ serde_json::to_writer(&mut w, value)?;
+ Ok(w.len())
+}
+
+/// Truncates `list` to fit within `payload_size_max_bytes` when serialized to
+/// JSON.
+pub fn shrink_to_fit<T: Serialize>(list: &mut Vec<T>, payload_size_max_bytes: usize) -> Result<()> {
+ let size = compute_serialized_size(&list)?;
+ // See bug 535326 comment 8 for an explanation of the estimation
+ match ((payload_size_max_bytes / 4) * 3).checked_sub(1500) {
+ Some(max_serialized_size) => {
+ if size > max_serialized_size {
+ // Estimate a little more than the direct fraction to maximize packing
+ let cutoff = (list.len() * max_serialized_size - 1) / size + 1;
+ list.truncate(cutoff + 1);
+ // Keep dropping off the last entry until the data fits.
+ while compute_serialized_size(&list)? > max_serialized_size {
+ if list.pop().is_none() {
+ break;
+ }
+ }
+ }
+ Ok(())
+ }
+ None => {
+ list.clear();
+ Ok(())
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::clients::record::CommandRecord;
+
+ #[test]
+ fn test_compute_serialized_size() {
+ assert_eq!(compute_serialized_size(&1).unwrap(), 1);
+ assert_eq!(compute_serialized_size(&"hi").unwrap(), 4);
+ assert_eq!(
+ compute_serialized_size(&["hi", "hello", "bye"]).unwrap(),
+ 20
+ );
+ }
+
+ #[test]
+ fn test_shrink_to_fit() {
+ let mut commands = vec![
+ CommandRecord {
+ name: "wipeEngine".into(),
+ args: vec!["bookmarks".into()],
+ flow_id: Some("flow".into()),
+ },
+ CommandRecord {
+ name: "resetEngine".into(),
+ args: vec!["history".into()],
+ flow_id: Some("flow".into()),
+ },
+ CommandRecord {
+ name: "logout".into(),
+ args: Vec::new(),
+ flow_id: None,
+ },
+ ];
+
+ // 4096 bytes is enough to fit all three commands.
+ shrink_to_fit(&mut commands, 4096).unwrap();
+ assert_eq!(commands.len(), 3);
+
+ let sizes = commands
+ .iter()
+ .map(|c| compute_serialized_size(c).unwrap())
+ .collect::<Vec<_>>();
+ assert_eq!(sizes, &[61, 60, 30]);
+
+ // `logout` won't fit within 2168 bytes.
+ shrink_to_fit(&mut commands, 2168).unwrap();
+ assert_eq!(commands.len(), 2);
+
+ // `resetEngine` won't fit within 2084 bytes.
+ shrink_to_fit(&mut commands, 2084).unwrap();
+ assert_eq!(commands.len(), 1);
+
+ // `wipeEngine` won't fit at all.
+ shrink_to_fit(&mut commands, 1024).unwrap();
+ assert!(commands.is_empty());
+ }
+}