Fix for state mutation through CLI

Reported by @pcholakov, state mutations fail in 0.9.0. This fixes two issues:
- Propagation of the error back to the CLI
- Serde HashMap<Bytes, Bytes> as Seq to get around flexbuffer's limitation

Note: this needs to be cherry picked to 0.9 release.

Closes #1486

cli/src/clients/metas_client.rs

@@ -164,8 +164,9 @@ impl MetasClient {
         T: DeserializeOwned + Send,
     {
         debug!("Sending request {} ({})", method, path);
-        let request = self.prepare(method, path);
+        let request = self.prepare(method, path.clone());
         let resp = request.send().await?;
+        debug!("Response from {} ({})", path, resp.status());
         Ok(resp.into())
     }
 
@@ -180,8 +181,9 @@ impl MetasClient {
         B: Serialize + std::fmt::Debug + Send,
     {
         debug!("Sending request {} ({}): {:?}", method, path, body);
-        let request = self.prepare_with_body(method, path, body);
+        let request = self.prepare_with_body(method, path.clone(), body);
         let resp = request.send().await?;
+        debug!("Response from {} ({})", path, resp.status());
         Ok(resp.into())
     }
 }

cli/src/commands/state/util.rs

@@ -70,7 +70,7 @@ pub(crate) async fn update_state(
     };
 
     let client = MetasClient::new(env)?;
-    client.patch_state(service, req).await.unwrap();
+    let _ = client.patch_state(service, req).await?.success_or_error()?;
 
     Ok(())
 }

crates/types/src/state_mut.rs

@@ -8,20 +8,26 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-use crate::identifiers::ServiceId;
+use std::collections::HashMap;
+use std::fmt::{Display, Formatter};
+
 use base64::Engine;
 use bytes::Bytes;
+use serde_with::serde_as;
 use sha2::{Digest, Sha256};
-use std::collections::HashMap;
-use std::fmt::{Display, Formatter};
 
+use crate::identifiers::ServiceId;
+
+#[serde_as]
 /// ExternalStateMutation
 ///
 /// represents an external request to mutate a user's state.
 #[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
 pub struct ExternalStateMutation {
     pub service_id: ServiceId,
     pub version: Option<String>,
+    // flexbuffers only supports string-keyed maps :-( --> so we store it as vector of kv pairs
+    #[serde_as(as = "serde_with::Seq<(_, _)>")]
     pub state: HashMap<Bytes, Bytes>,
 }