Current Behavior
AsyncCurrentValueSubject may miss a value when a new consumer is added
concurrently with send.
In handleNewConsumer(), the current value is read from the state
before the new channel is registered:
let (terminalState, current) = state.withCriticalRegion {
($0.terminalState, $0.current)
}
...
asyncBufferedChannel.send(current)
...
state.withCriticalRegion {
$0.channels[consumerId] = asyncBufferedChannel
}
Since reading current and registering the channel are separate
operations, send can interleave between them.
For example, assuming the current value is A:
Consumer Producer
read current = A
unlock
send(B)
current = B
// new consumer isn't registered
channel.send(A)
register channel
The consumer receives A, but never receives B, even though B was
sent after the consumer started subscribing.
Unless another value is sent, the consumer is now observing a stale
value while the subject's current value is B.
There appears to be a similar race with termination: the subject can
terminate after terminalState is checked but before the new channel is
registered, potentially leaving the newly registered consumer unaware
that the subject has already terminated.
Expected Behavior
Establishing a new subscription should be linearizable with respect to
send and termination.
For a concurrent subscription and send(B), either:
subscribe -> send(B)
consumer receives: A, B
or:
send(B) -> subscribe
consumer receives: B
should be possible, but the newly subscribed consumer should not
permanently miss B.
Current Behavior
AsyncCurrentValueSubjectmay miss a value when a new consumer is addedconcurrently with
send.In
handleNewConsumer(), the current value is read from the statebefore the new channel is registered:
Since reading
currentand registering the channel are separateoperations,
sendcan interleave between them.For example, assuming the current value is
A:The consumer receives
A, but never receivesB, even thoughBwassent after the consumer started subscribing.
Unless another value is sent, the consumer is now observing a stale
value while the subject's current value is
B.There appears to be a similar race with termination: the subject can
terminate after
terminalStateis checked but before the new channel isregistered, potentially leaving the newly registered consumer unaware
that the subject has already terminated.
Expected Behavior
Establishing a new subscription should be linearizable with respect to
sendand termination.For a concurrent subscription and
send(B), either:or:
should be possible, but the newly subscribed consumer should not
permanently miss
B.