[FlatMapLatest] Fix crash and hang under a single consumer

[kSibalic]

Motivation

Consuming a flatMapLatest sequence with a single for await loop, where the base switches inner sequences frequently, can trap (#437):

FlatMapLatestStateMachine.swift:132: Precondition failed: Already have downstream continuation

The same scenario also intermittently hangs; the downstream continuation is never resumed.

There are two root causes:

• Unlocked state machine mutation: FlatMapLatestStorage.startInnerTask calls stateMachine.innerTaskStarted(_:generation:) without holding the lock, every path to it runs after lock.withLock has returned. This "read-modify-write" of the whole state tuple races with locked transitions: a stale write-back can resurrect an already resumed downstream continuation (the trap) or erase a freshly stored one (the hang). It also allows an empty inner sequence to run innerFinished before the task is registered, leaving the state claiming an inner is running forever.

• Non-atomic suspension: FlatMapLatestStorage.next() releases the lock before suspend() re-acquires it to call next(for:), so producers can legitimately buffer elements or finish inside that window. The preconditions in next(for:) assumed atomicity that doesn't hold there.

Modifications

• FlatMapLatestStorage.startInnerTask now holds the lock across task creation and registration. The task body's first action is acquiring that same lock, so it cannot transition the state machine until registration completes; the discipline startOuterTask already follows. After this change, every state machine access happens with the lock held.
• FlatMapLatestStateMachine.next(for:) serves the demand from the state it observes instead of trapping it pops a buffered element (or error), or finishes, before parking the continuation mirroring next(). The single-consumer precondition (downstreamCont == nil) remains, as it is a true invariant once registration is locked.
• Added a stress regression test based on the issue's reproduction, run 1000 times under a timeout so the hang variant fails the test instead of wedging the run.

Result

The regression test doubles as the reproduction. Checking out only the test file from this branch onto main, with the sources left unfixed, makes it fail in one of the two reported ways: a trap on the next(for:) precondition ("Already have downstream continuation"), or a 120-second timeout because the downstream continuation was lost and the consumer is never resumed.

Pre-fix failure output

Test Case '-[AsyncAlgorithmsTests.TestFlatMapLatest test_rapid_switching_single_consumer_stress]' started.
Tests/AsyncAlgorithmsTests/TestFlatMapLatest.swift:263: error: -[AsyncAlgorithmsTests.TestFlatMapLatest test_rapid_switching_single_consumer_stress] : Asynchronous wait failed: Exceeded timeout of 120 seconds, with unfulfilled expectations: "all iterations finished".
Test Case '-[AsyncAlgorithmsTests.TestFlatMapLatest test_rapid_switching_single_consumer_stress]' failed (120.283 seconds).

With this change, the same test passes in about 2.5 seconds. The hang disappears because the inner task is now registered under the lock, so no stale write-back can erase a stored downstream continuation. The trap disappears for the same reason (no write-back can resurrect an already resumed continuation) and because next(for:) now serves elements buffered during the legitimate unlock window instead of asserting that the window was empty.

Resolves #437.