Updates to convex-rs

GitOrigin-RevId: b27f3b383f2a8a1f84d558c33c5b327ad2b94b13

Author: Convex, Inc

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Upcoming
 
+# 0.6.0
+
+- Remove support for Set and Map Convex types. These types are deprecated.
+- Add comprehensive support for ConvexError with `data` payload as part of the
+ `FunctionResult` enum.
+- Better support for emitting loglines
+
 # 0.5.0
 
 - Prelim support for ConvexError, encoded into an anyhow::Error. Eventual plan

Cargo.toml

@@ -2,7 +2,7 @@
 name = "convex"
 description = "Client library for Convex (convex.dev)"
 authors = [ "Convex, Inc. <team@convex.dev>" ]
-version = "0.5.0"
+version = "0.6.0"
 edition = "2021"
 rust-version = "1.65.0"
 resolver = "2"
@@ -15,26 +15,25 @@ anyhow = { version = "1" }
 async-trait = { version = "0.1" }
 base64 = { version = "0.13" }
 bytes = { version = "1.1.0" }
-convex_sync_types = { path = "./sync_types", version = "=0.5.0" }
-derive_more = { version = "0.99" }
+convex_sync_types = { path = "./sync_types", version = "=0.6.0" }
 futures = { version = "0.3" }
 imbl = { version = "2.0.0" }
 parking_lot = { optional = true, version = "0.12" }
 proptest = { optional = true, version = "1" }
 proptest-derive = { optional = true, version = "0.4.0" }
 rand = { version = "0.8" }
-serde = { features = [ "derive" ], version = "1" }
 serde_json = { features = [ "float_roundtrip", "preserve_order" ], version = "1" }
+thiserror = { version = "1" }
 tokio = { features = [ "full" ], version = "1" }
 tokio-stream = { features = [ "io-util", "sync" ], version = "^0.1.8" }
-tokio-tungstenite = { version = "0.18.0" }
+tokio-tungstenite = { version = "0.20.0" }
 tracing = { version = "0.1" }
 url = { version = "2" }
-uuid = { features = [ "serde", "v4" ], version = "1.4" }
+uuid = { features = [ "serde", "v4" ], version = "1.6" }
 
 [dev-dependencies]
 colored = { version = "2" }
-convex_sync_types = { path = "./sync_types", version = "=0.5.0", features = [ "testing" ] }
+convex_sync_types = { path = "./sync_types", version = "=0.6.0", features = [ "testing" ] }
 dotenvy = { version = "0.15.7" }
 maplit = { version = "1" }
 parking_lot = { version = "0.12" }
@@ -49,4 +48,4 @@ native-tls = [ "tokio-tungstenite/native-tls" ]
 native-tls-vendored = [ "tokio-tungstenite/native-tls-vendored" ]
 rustls-tls-native-roots = [ "tokio-tungstenite/rustls-tls-native-roots" ]
 rustls-tls-webpki-roots = [ "tokio-tungstenite/rustls-tls-webpki-roots" ]
-testing = [ "proptest", "proptest-derive", "parking_lot" ]
+testing = [ "convex_sync_types/testing", "proptest", "proptest-derive", "parking_lot" ]

README.md

@@ -56,3 +56,8 @@ tracing_subscriber::fmt()
  .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
  .init();
 ```
+
+By default, this will emit all logs, including internal logs from the client.
+Logs from your Convex backend will show up under the `convex_logs` target at
+Level=DEBUG. If you want to isolate just those logs, please refer to the
+[tracing_subscriber documentation](https://docs.rs/tracing-subscriber/latest/tracing_subscriber/layer/index.html#filtering-with-layers).

examples/convex_chat_client.rs

@@ -162,6 +162,9 @@ async fn main() -> anyhow::Result<()> {
  FunctionResult::ErrorMessage(err) => {
  println!("{}.", err.red().bold());
  },
+ FunctionResult::ConvexError(err) => {
+ println!("{:?}", err);
+ },
  };
  }
 

examples/quickstart/package.json

@@ -1,5 +1,5 @@
 {
  "dependencies": {
- "convex": "^0.13.0"
+ "convex": "1.8.0"
  }
 }

src/base_client/mod.rs

@@ -32,11 +32,13 @@ use tokio::sync::oneshot;
 #[cfg(doc)]
 use crate::ConvexClient;
 use crate::{
+ convex_logs,
  sync::{
  ReconnectProtocolReason,
  ServerMessage,
  },
  value::Value,
+ ConvexError,
 };
 
 mod request_manager;
@@ -71,7 +73,7 @@ struct Query {
 }
 
 /// An identifier for a single subscriber to a query.
-#[derive(Copy, Clone, Debug, Default, Eq, PartialEq, PartialOrd, Ord)]
+#[derive(Copy, Clone, Debug, Default, Eq, PartialEq, PartialOrd, Ord, Hash)]
 #[cfg_attr(test, derive(proptest_derive::Arbitrary))]
 pub struct SubscriberId(QueryId, usize);
 
@@ -165,7 +167,7 @@ impl LocalSyncState {
  None => panic!("INTERNAL BUG: Unknown query id {query_id}"),
  Some(t) => t,
  };
- let mut local_query = match self.query_set.get_mut(&query_token) {
+ let local_query = match self.query_set.get_mut(&query_token) {
  None => panic!("INTERNAL BUG: No query found for query token {query_token:?}",),
  Some(q) => q,
  };
@@ -273,7 +275,8 @@ impl RemoteQuerySet {
  start_version,
  end_version,
  modifications,
- } = transition else {
+ } = transition
+ else {
  panic!("not transition");
  };
  if start_version != self.version {
@@ -290,22 +293,33 @@ impl RemoteQuerySet {
  StateModification::QueryUpdated {
  query_id,
  value,
- log_lines: _,
+ log_lines,
  journal: _,
  } => {
+ for log_line in log_lines.0 {
+ convex_logs!("{}", log_line);
+ }
  self.remote_query_set
  .insert(query_id, FunctionResult::Value(value));
  },
  StateModification::QueryFailed {
  query_id,
  error_message,
- log_lines: _,
+ log_lines,
  journal: _,
- // TODO @srb: Implement ConvexError in Rust client queries
- error_data: _,
+ error_data,
  } => {
- self.remote_query_set
- .insert(query_id, FunctionResult::ErrorMessage(error_message));
+ for log_line in log_lines.0 {
+ convex_logs!("{}", log_line);
+ }
+ let function_result = match error_data {
+ Some(v) => FunctionResult::ConvexError(ConvexError {
+ message: error_message,
+ data: v,
+ }),
+ None => FunctionResult::ErrorMessage(error_message),
+ };
+ self.remote_query_set.insert(query_id, function_result);
  },
  StateModification::QueryRemoved { query_id } => {
  self.remote_query_set.remove(&query_id);
@@ -368,7 +382,7 @@ impl OptimisticQueryResults {
 /// The main methods, [`subscribe`](Self::subscribe()),
 /// [`unsubscribe`](Self::unsubscribe()), and
 /// [`mutation`](Self::mutation()) directly correspond to its
-/// equivalent for the external [ConvexClient](crate::ConvexClient).
+/// equivalent for the external [ConvexClient].
 ///
 /// The only different method is [`get_query`](Self::get_query()), which
 /// returns the current value for a query given its query id. This method can be
@@ -583,22 +597,16 @@ impl BaseConvexClient {
  }
  return Ok(Some(self.state.latest_results.clone()));
  },
- ServerMessage::QueriesFailed { failures } => {
- // Note that we never expect to receive this as it is not sent by the server.
- for failure in failures {
- tracing::error![failure.message];
- }
- tracing::error!(
- "Received unexpected QueriesFailed from server. Restarting protocol."
- );
- return Err("QueriesFailed, see tracing::error for more details.".to_string());
- },
  ServerMessage::MutationResponse {
  request_id,
  result,
  ts,
- log_lines: _,
+ log_lines,
  } => {
+ for log_line in log_lines.0 {
+ convex_logs!("{}", log_line);
+ }
+
  if let Some(ts) = ts {
  self.observe_timestamp(ts);
  }
@@ -629,8 +637,11 @@ impl BaseConvexClient {
  ServerMessage::ActionResponse {
  request_id,
  result,
- log_lines: _,
+ log_lines,
  } => {
+ for log_line in log_lines.0 {
+ convex_logs!("{}", log_line);
+ }
  let request_id = RequestId::new(request_id);
  self.request_manager.update_request(
  &request_id,
@@ -672,9 +683,7 @@ impl BaseConvexClient {
  let remote_query_results = &self.remote_query_set.remote_query_set;
  let mut query_id_to_value = BTreeMap::new();
  for (query_id, result) in remote_query_results.iter() {
- let Some(_udf_path) = self
- .state
- .query_path(*query_id) else {
+ let Some(_udf_path) = self.state.query_path(*query_id) else {
  // It's possible that we've already unsubscribed to this query but
  // the server hasn't learned about that yet. If so, ignore this one.
  continue;
@@ -701,3 +710,17 @@ impl BaseConvexClient {
  self.optimistic_query_results.query_result(query_id)
  }
 }
+
+/// Macro used for piping UDF logs to a custom formatter that exposes
+/// just the log content, without any additional Rust metadata.
+#[macro_export]
+macro_rules! convex_logs {
+ (target: $target:expr, $($arg:tt)+) => {
+ tracing::event!(target: "convex_logs", tracing::Level::DEBUG, $($arg)+);
+ // Additional custom behavior can be added here
+ };
+ ($($arg:tt)+) => {
+ tracing::event!(target: "convex_logs", tracing::Level::DEBUG, $($arg)+);
+ // Additional custom behavior can be added here
+ };
+}

src/base_client/query_result.rs

@@ -8,7 +8,10 @@ use imbl::{
 };
 
 use super::SubscriberId;
-use crate::Value;
+use crate::{
+ ConvexError,
+ Value,
+};
 
 /// Result of a Convex function (query/mutation/action).
 ///
@@ -20,14 +23,19 @@ pub enum FunctionResult {
  /// The error message of a Convex function run that does not complete
  /// successfully.
  ErrorMessage(String),
+ /// The error payload of a Convex function run that doesn't complete
+ /// successfully, with an application-level error.
+ ConvexError(ConvexError),
 }
 
 impl From<Result<Value, ErrorPayload<Value>>> for FunctionResult {
  fn from(result: Result<Value, ErrorPayload<Value>>) -> Self {
  match result {
  Ok(value) => FunctionResult::Value(value),
- // TODO @srb: Implement ConvexError in Rust client
- Err(error) => FunctionResult::ErrorMessage(error.get_message().to_owned()),
+ Err(ErrorPayload::ErrorData { message, data }) => {
+ FunctionResult::ConvexError(ConvexError { message, data })
+ },
+ Err(ErrorPayload::Message(message)) => FunctionResult::ErrorMessage(message),
  }
  }
 }
@@ -36,8 +44,11 @@ impl From<FunctionResult> for Result<Value, ErrorPayload<Value>> {
  fn from(result: FunctionResult) -> Self {
  match result {
  FunctionResult::Value(value) => Ok(value),
- // TODO @srb: Implement ConvexError in Rust client
  FunctionResult::ErrorMessage(error) => Err(ErrorPayload::Message(error)),
+ FunctionResult::ConvexError(error) => Err(ErrorPayload::ErrorData {
+ message: error.message,
+ data: error.data,
+ }),
  }
  }
 }
@@ -47,6 +58,9 @@ impl std::fmt::Debug for FunctionResult {
  match self {
  FunctionResult::Value(value) => f.debug_tuple("Value").field(value).finish(),
  FunctionResult::ErrorMessage(error) => write!(f, "{error}"),
+ FunctionResult::ConvexError(error) => {
+ f.debug_tuple("ConvexError").field(error).finish()
+ },
  }
  }
 }

src/lib.rs

@@ -44,7 +44,10 @@
 mod value;
 #[cfg(any(test, feature = "testing"))]
 pub use value::export::roundtrip::ExportContext;
-pub use value::Value;
+pub use value::{
+ ConvexError,
+ Value,
+};
 
 mod client;
 pub use client::{

src/sync/web_socket_manager.rs

@@ -229,7 +229,7 @@ impl WebSocketWorker {
  let server_message = json.try_into()?;
  match server_message {
  ServerMessage::Ping => tracing::trace!("received message {server_message:?}"),
- _ => tracing::debug!("received message {server_message:?}"),
+ _ => tracing::trace!("received message {server_message:?}"),
  };
 
  let resp = ProtocolResponse::ServerMessage(server_message);

src/value/export/mod.rs

@@ -53,8 +53,6 @@ impl Value {
  Value::Array(values) => {
  JsonValue::Array(values.into_iter().map(|x| x.export()).collect())
  },
- // Use the internal representation for deprecated types
- Value::Set(_) | Value::Map(_) => self.into(),
  Value::Object(map) => JsonValue::Object(
  map.into_iter()
  .map(|(key, value)| (key, value.export()))
@@ -66,10 +64,7 @@ impl Value {
 
 #[cfg(test)]
 mod tests {
- use maplit::{
- btreemap,
- btreeset,
- };
+ use maplit::btreemap;
  use serde_json::json;
 
  use super::*;
@@ -193,42 +188,4 @@ mod tests {
  }),
  );
  }
-
- #[test]
- fn maps_use_the_internal_representation() {
- assert_eq!(
- Value::Map(btreemap! {
- "a".into() => 1.into(),
- "b".into() => 2.into(),
- "c".into() => 3.into(),
- })
- .export(),
- json!({
- "$map": [
- ["a", { "$integer": "AQAAAAAAAAA=" }],
- ["b", { "$integer": "AgAAAAAAAAA=" }],
- ["c", { "$integer": "AwAAAAAAAAA=" }],
- ],
- }),
- );
- }
-
- #[test]
- fn sets_use_the_internal_representation() {
- assert_eq!(
- Value::Set(btreeset! {
- 1.into(),
- 2.into(),
- 3.into(),
- })
- .export(),
- json!({
- "$set": [
- { "$integer": "AQAAAAAAAAA=" },
- { "$integer": "AgAAAAAAAAA=" },
- { "$integer": "AwAAAAAAAAA=" },
- ],
- }),
- );
- }
 }