fix: unsubscribe panics (#4938)

# Description of Changes

This PR updates Rust SDK subscription unsubscribe handling to return an
error instead of panicking when the internal pending mutation channel is
closed.

## Motivation

`SubscriptionHandle::unsubscribe_then` currently calls `unwrap()` after
sending an unsubscribe mutation through the internal pending mutation
channel. If the connection or subscription manager has already shut
down, that send can fail with a disconnected channel error, causing the
application to panic.

I encountered this in [`bevy_stdb`](https://github.com/onx2/bevy_stdb),
which tracks subscriptions across reconnects and unsubscribes old
handles after applying a new subscription for the same key.

## Changes

- Replaces `unwrap()` on `pending_mutation_sender.unbounded_send(...)`
with error propagation.
- Returns `crate::Error::Internal(...)` when the unsubscribe mutation
cannot be sent.
- Updates local unsubscribe state only after the unsubscribe mutation is
successfully queued.

# API and ABI breaking changes

None expected.

This changes an internal error path from panicking to returning an
existing `crate::Error::Internal(...)` through the existing
`crate::Result<()>` return type.

# Expected complexity level and risk

Complexity: 1/5

This is a small, localized change. The method already returns
`crate::Result<()>`, so replacing the `unwrap()` with error propagation
fits the existing API shape.

Risk is low. Callers now receive an error if the internal pending
mutation channel is closed instead of panicking. Existing `AlreadyEnded`
and `AlreadyUnsubscribed` behavior is unchanged.

# Testing

- [x] Verified that `unsubscribe_then` returns
`Err(crate::Error::Internal(...))` instead of panicking when the pending
mutation channel is disconnected.
- [ ] Verified that successful unsubscribe behavior is unchanged.
- [ ] Verified that `AlreadyEnded` is still returned when the
subscription has already ended.
- [ ] Verified that `AlreadyUnsubscribed` is still returned when
unsubscribe was already requested.

---------

Signed-off-by: Jeff Rooks <onx2rj@gmail.com>

sdks/rust/src/subscription.rs

@@ -379,21 +379,21 @@ impl<M: SpacetimeModule> SubscriptionState<M> {
         if self.is_ended() {
             return Err(crate::Error::AlreadyEnded);
         }
-        // Check if it has already been called.
+
         if self.unsubscribe_called {
             return Err(crate::Error::AlreadyUnsubscribed);
         }
 
-        self.unsubscribe_called = true;
-        self.on_ended = on_end;
-        // self.on_ended = Some(Box::new(on_end));
-
         // We send this even if the status is still Pending, so we can remove it from the manager.
         self.pending_mutation_sender
             .unbounded_send(PendingMutation::Unsubscribe {
                 query_set_id: self.query_set_id,
             })
-            .unwrap();
+            .map_err(|_| crate::Error::AlreadyEnded)?;
+
+        self.unsubscribe_called = true;
+        self.on_ended = on_end;
+
         Ok(())
     }