Count sync worker mutation channel full as rate limited (#28450)

If mutation channel is full, then client has sent too many mutations.
Classify as 429 rather than 503 at this layer.

GitOrigin-RevId: 4333d2248957056effb8d26230e6fabd1fecfb80

Author: nipunn1313

crates/application/src/application_function_runner/mod.rs

@@ -424,7 +424,11 @@ impl Limiter {
                 log_function_wait_timeout(self.env, self.udf_type);
                 anyhow::bail!(ErrorMetadata::rate_limited(
                     "TooManyConcurrentRequests",
-                    "Too many concurrent requests, backoff and try again.",
+                    format!(
+                        "Too many concurrent requests. Your backend is limited to {} concurrent {}s. To get more resources, upgrade to Convex Pro. If you are already on Convex Pro, please contact support.",
+                        self.total_permits,
+                        self.udf_type.to_lowercase_string(),
+                    ),
                 ));
             },
         }

crates/errors/src/lib.rs

@@ -153,6 +153,11 @@ impl ErrorMetadata {
     /// The short_msg should be a CapitalCamelCased describing the error (eg
     /// TooManyTeams). The msg should be a descriptive message targeted
     /// toward the developer.
+    ///
+    /// In the user facing `msg`, be very clear about what is actionable here.
+    /// Some rate limits require paying for more resources, while others are
+    /// indicative of incorrect user behavior (eg > 1000 concurrent mutations
+    /// over a single websocket).
     pub fn rate_limited(
         short_msg: impl Into<Cow<'static, str>>,
         msg: impl Into<Cow<'static, str>>,

crates/sync/src/worker.rs

@@ -521,9 +521,13 @@ impl<RT: Runtime> SyncWorker<RT> {
                 .boxed();
                 self.mutation_sender.try_send(future).map_err(|err| {
                     if err.is_full() {
-                        anyhow::anyhow!(ErrorMetadata::overloaded(
+                        anyhow::anyhow!(ErrorMetadata::rate_limited(
                             "TooManyConcurrentMutations",
-                            "Too many concurrent mutations, backoff and try again.",
+                            format!(
+                                "Too many concurrent mutations. Only up to \
+                                 {OPERATION_QUEUE_BUFFER_SIZE} pending mutations allowed on a \
+                                 single websocket."
+                            ),
                         ))
                     } else {
                         anyhow::anyhow!("Failed to send to mutation channel: {err}")