fix(subscription): don't panic when subscribing during an emit - #1105

The bug

SubscriberSet::retain checks an emitter's subscriber map out and leaves a ThreadId marker in its place while it invokes callbacks. SubscriberSet::insert unwraps that slot as if it always held the map:

lock.subscribers
    .entry(emitter_key_1)
    .or_insert_with(|| Either::Left(BTreeMap::new()))
    .as_mut()
    .unwrap_left()   // panics when the slot holds Right(ThreadId)

So subscribing to an emitter that is mid-emit panics:

called `Either::unwrap_left()` on a `Right` value: ThreadId(7)

This is reachable whenever one thread subscribes while another emits for the same key, or a callback subscribes re-entrantly. retain itself handles contention (it sees Right, drops the lock, sleeps, retries) — only insert assumes the map is always there.

We hit this in a server that holds a LoroDoc per document: a subscribe_local_update racing an import panicked the tokio worker. The process stayed up and the socket stayed open, so clients waited on a response that never came — one occurrence wedged a test run for eight hours before we noticed.

The fix

The marker can't simply be replaced with a fresh map: it is what makes a concurrent retain wait rather than emit the same emitter twice, and what lets is_recursive_calling detect re-entrancy.

So subscribers added during an emit are parked in a new pending_subscribers map and folded back in exactly where retain already merges the ones added while the callback ran:

// Add any new subscribers that were added while invoking the callback.
if let Some(Either::Left(new_subscribers)) = lock.subscribers.remove(emitter) {
    subscribers.extend(new_subscribers);
}
if let Some(parked) = lock.pending_subscribers.remove(emitter) {
    subscribers.extend(parked);
}

Merged before the dropped sweep, so a subscriber unsubscribed again mid-emit is still dropped. remove drops parked subscribers along with the emitter they belong to.

Semantics: a subscriber added mid-emit does not receive the event already being delivered, and is registered for later ones. That matches what the existing merge does for the case it already handled.

Tests

Two, both in crates/loro-internal/src/utils/subscription.rs:

• insert_during_emit_does_not_panic — a callback subscribes re-entrantly; asserts the new subscriber misses the in-flight event and receives the next one.
• concurrent_insert_and_emit_do_not_panic — one thread parks inside a callback (emitter checked out) while another subscribes to the same key.

Both reproduce the original panic verbatim when the insert change is reverted. The rest of loro-internal's suite (380 tests) passes.