From 2abdd69d770e9f2375a56bf56d3a49754be95f99 Mon Sep 17 00:00:00 2001 From: Max Heimbrock <43608204+MaxHeimbrock@users.noreply.github.com> Date: Mon, 20 Jul 2026 12:32:43 +0200 Subject: [PATCH 1/4] Increased test coverage to test for eos and chunk burst --- .../DataStreamIncrementalReadTests.cs | 129 ++++++++++++++++-- Tests/PlayMode/UniTask/StreamUniTaskTests.cs | 26 +++- 2 files changed, 141 insertions(+), 14 deletions(-) diff --git a/Tests/EditMode/DataStreamIncrementalReadTests.cs b/Tests/EditMode/DataStreamIncrementalReadTests.cs index 0129f6d0..f3aba861 100644 --- a/Tests/EditMode/DataStreamIncrementalReadTests.cs +++ b/Tests/EditMode/DataStreamIncrementalReadTests.cs @@ -86,12 +86,11 @@ public void OnChunk_ConcurrentProducerAndConsumer_AllChunksObservedInOrder() Assert.AreEqual(i.ToString(), observed[i], $"Chunk reordering at index {i}."); } - // OnEos races with OnChunk on the FFI thread. After both complete, the lock - // must leave the instruction in a consistent state: IsEos visible, no torn - // reads of LatestChunk, and the chunks pushed before EOS still drainable - // (up to the point where Reset() is correctly disallowed past EOS). + // OnEos races with OnChunk on the FFI thread. EOS is an ordered item behind the + // chunks, so after both complete the consumer must be able to drain every chunk in + // FIFO order before observing end-of-stream — never seeing EOS while data remains. [Test] - public void OnEos_RacingWithChunks_FinalStateConsistent() + public void OnEos_RacingWithChunks_AllChunksDrainThenEos() { for (int trial = 0; trial < 50; trial++) { @@ -109,20 +108,128 @@ public void OnEos_RacingWithChunks_FinalStateConsistent() start.Set(); producer.Wait(TimeSpan.FromSeconds(5)); - Assert.IsTrue(reader.IsEos, $"Trial {trial}: EOS not observed after producer finished."); - // Reading LatestChunk after EOS must never throw an unexpected exception - // — the only allowed throw is the StreamError on protocol error, which - // we don't simulate here. Specifically: no NullReferenceException, no - // collection-modified exception, no torn-read. + var observed = new List(100); + var deadline = DateTime.UtcNow.AddSeconds(5); + while (true) + { + while (reader.keepWaiting) + { + if (DateTime.UtcNow > deadline) + Assert.Fail($"Trial {trial}: drain stalled at {observed.Count}/100 — chunk lost or deadlock."); + Thread.Yield(); + } + if (reader.IsEos) break; + observed.Add(reader.Value); + reader.Reset(); + } + + Assert.AreEqual(100, observed.Count, + $"Trial {trial}: EOS observed before all chunks were drained."); + for (int i = 0; i < 100; i++) + Assert.AreEqual(i.ToString(), observed[i], $"Trial {trial}: chunk reordering at index {i}."); + + // A clean end-of-stream leaves no chunk to read: LatestChunk returns the + // default value and never throws (a torn read would surface here). Assert.DoesNotThrow(() => { var _ = reader.Value; }, $"Trial {trial}: reading LatestChunk after EOS race threw unexpectedly."); + Assert.IsNull(reader.Value, $"Trial {trial}: expected null chunk at end-of-stream."); + } + } + + // The whole stream — every chunk AND end-of-stream — arrives before the consumer + // reads anything (the common small-stream / already-buffered case). Every chunk must + // still be delivered before EOS is observed. This is the core regression the + // single-queue / EOS-as-terminal-item fix addresses: previously IsEos short-circuited + // the buffered chunks and the whole stream was dropped. + [Test] + public void Burst_AllChunksThenEosBeforeConsumer_AllDrainedThenEos() + { + using var handle = new FfiHandle(IntPtr.Zero); + var reader = new TestIncrementalReader(handle); + + reader.PushChunk("A"); + reader.PushChunk("B"); + reader.PushChunk("C"); + reader.PushEos(); + + CollectionAssert.AreEqual(new[] { "A", "B", "C" }, Drain(reader)); + Assert.IsTrue(reader.IsEos); + } + + // A single chunk followed immediately by end-of-stream — the previous implementation + // dropped the chunk because IsEos short-circuited the read on the first iteration. + [Test] + public void Burst_SingleChunkThenEos_ChunkDelivered() + { + using var handle = new FfiHandle(IntPtr.Zero); + var reader = new TestIncrementalReader(handle); + + reader.PushChunk("only"); + reader.PushEos(); + + CollectionAssert.AreEqual(new[] { "only" }, Drain(reader)); + Assert.IsTrue(reader.IsEos); + } + + // An empty stream (end-of-stream with no chunks) drains to zero chunks and reports EOS. + [Test] + public void Burst_EosWithNoChunks_ReportsEosWithNoChunks() + { + using var handle = new FfiHandle(IntPtr.Zero); + var reader = new TestIncrementalReader(handle); + + reader.PushEos(); + + CollectionAssert.IsEmpty(Drain(reader)); + Assert.IsTrue(reader.IsEos); + } + + // Chunks that precede an error close are still delivered in order; the error then + // surfaces via IsError/Error once the consumer reaches the terminal marker. + [Test] + public void Burst_ChunksThenErrorEos_ChunksDrainedThenErrorSurfaces() + { + using var handle = new FfiHandle(IntPtr.Zero); + var reader = new TestIncrementalReader(handle); + + reader.PushChunk("A"); + reader.PushChunk("B"); + reader.PushEos(new LiveKit.Proto.StreamError { Description = "boom" }); + + // Read-first drain; break on EOS before reading (an error terminal has no chunk, + // and reading it would rethrow the StreamError). + var observed = new List(); + while (!reader.keepWaiting) + { + if (reader.IsEos) break; + observed.Add(reader.Value); + reader.Reset(); + } + + CollectionAssert.AreEqual(new[] { "A", "B" }, observed); + Assert.IsTrue(reader.IsEos); + Assert.IsTrue(reader.IsError); + Assert.AreEqual("boom", reader.Error.Message); + } + + // Canonical read-first drain used by the tests above: spin until the reader is + // positioned on a chunk or the terminal marker, break on EOS, else read and advance. + private static List Drain(TestIncrementalReader reader) + { + var observed = new List(); + while (!reader.keepWaiting) + { + if (reader.IsEos) break; + observed.Add(reader.Value); + reader.Reset(); } + return observed; } // OnChunk-after-Reset is the most common interleaving in production: the // consumer's coroutine just yielded and called Reset, and an FFI thread // pushes a chunk before the consumer wakes. The lock must serialize them - // so the new chunk is correctly placed (either as _latestChunk if the + // so the new chunk is correctly placed (either as the current item if the // queue was empty, or appended to the queue). [Test] public void OnChunk_RacingWithReset_NoChunkLost() diff --git a/Tests/PlayMode/UniTask/StreamUniTaskTests.cs b/Tests/PlayMode/UniTask/StreamUniTaskTests.cs index eaa067eb..84f05f80 100644 --- a/Tests/PlayMode/UniTask/StreamUniTaskTests.cs +++ b/Tests/PlayMode/UniTask/StreamUniTaskTests.cs @@ -55,9 +55,8 @@ public System.Collections.IEnumerator AsAsyncEnumerable_DeliversChunksInOrder_Th } }); - // The current chunk is delivered even when EoS is already set at the time it is read, - // then the sequence ends. (Chunks buffered beyond the current one when EoS arrives are - // not drainable, because the reader disallows Reset() past EoS.) + // The final chunk is delivered even when EoS is already set at the time it is read, + // then the sequence ends. [UnityTest] public System.Collections.IEnumerator AsAsyncEnumerable_DeliversFinalChunkThenEos() => UniTask.ToCoroutine(async () => { @@ -74,6 +73,27 @@ public System.Collections.IEnumerator AsAsyncEnumerable_DeliversFinalChunkThenEo CollectionAssert.AreEqual(new[] { "only" }, observed); }); + // Every chunk buffered before EoS is drained, even when all of them plus EoS arrive + // before the consumer reads anything (the already-buffered burst case). EoS is an + // ordered item at the tail of the chunk queue, so it can never short-circuit the data. + [UnityTest] + public System.Collections.IEnumerator AsAsyncEnumerable_DrainsAllBufferedChunks_WhenEosArrivesFirst() => UniTask.ToCoroutine(async () => + { + using var handle = new FfiHandle(IntPtr.Zero); + var reader = new TestIncrementalReader(handle); + + reader.PushChunk("A"); + reader.PushChunk("B"); + reader.PushChunk("C"); + reader.PushEos(); + + var observed = new List(); + await foreach (var chunk in reader.AsAsyncEnumerable()) + observed.Add(chunk); + + CollectionAssert.AreEqual(new[] { "A", "B", "C" }, observed); + }); + // A chunk delivered before the stream errors is observed; the subsequent error EoS // then surfaces as a thrown StreamError. Manual enumeration models the real timeline // (chunk arrives, is consumed, THEN the error ends the stream) — note that once the From d2522f57f04711148f68560369b6874171de1c02 Mon Sep 17 00:00:00 2001 From: Max Heimbrock <43608204+MaxHeimbrock@users.noreply.github.com> Date: Mon, 20 Jul 2026 12:32:21 +0200 Subject: [PATCH 2/4] Fix for eos and stream queue --- Runtime/Scripts/DataStreams/ByteDataStream.cs | 5 +- Runtime/Scripts/DataStreams/DataStream.cs | 117 +++++++++++++----- Runtime/Scripts/DataStreams/TextDataStream.cs | 5 +- .../UniTask/StreamReaderUniTaskExtensions.cs | 38 ++---- .../Runtime/Agent/TranscriptionReader.cs | 14 +-- 5 files changed, 107 insertions(+), 72 deletions(-) diff --git a/Runtime/Scripts/DataStreams/ByteDataStream.cs b/Runtime/Scripts/DataStreams/ByteDataStream.cs index 3823a976..73d6c971 100644 --- a/Runtime/Scripts/DataStreams/ByteDataStream.cs +++ b/Runtime/Scripts/DataStreams/ByteDataStream.cs @@ -145,8 +145,9 @@ internal ReadAllInstruction(ulong asyncId) /// YieldInstruction for . /// /// - /// Usage: while is false (i.e. the stream has not ended), - /// call , yield the instruction, and then access . + /// Usage: loop yielding the instruction; when it resumes, break if + /// is true (the stream has ended and every chunk has been drained), otherwise read + /// and call to advance to the next chunk. /// public sealed class ReadIncrementalInstruction : ReadIncrementalInstructionBase { diff --git a/Runtime/Scripts/DataStreams/DataStream.cs b/Runtime/Scripts/DataStreams/DataStream.cs index 68ee95a9..dd88b469 100644 --- a/Runtime/Scripts/DataStreams/DataStream.cs +++ b/Runtime/Scripts/DataStreams/DataStream.cs @@ -72,19 +72,44 @@ internal StreamError(Proto.StreamError proto) : base(proto.Description) { } /// /// Shared state and helpers for incremental stream reader yield instructions. - /// Holds the latest chunk, end-of-stream flag, and error; subclasses own the - /// typed event subscription and convert raw event payloads via - /// and . + /// Models the incoming stream as a single ordered queue whose items are either a + /// chunk or a terminal end-of-stream marker (carrying an optional ). + /// The consumer advances through the queue one item at a time via ; + /// end-of-stream is only observed once every buffered chunk has been drained, mirroring + /// the JS and Python SDKs. Subclasses own the typed event subscription and convert raw + /// event payloads via and . /// public abstract class ReadIncrementalInstructionBase : StreamYieldInstruction { private readonly ulong _handleValue; - private readonly Queue _pendingChunks = new(); - private TContent _latestChunk; - // Chunk events arrive on the FFI thread; Reset() and the LatestChunk getter - // run on the main-thread coroutine. _gate serializes mutations of the queue, - // _latestChunk, IsCurrentReadDone, IsEos, and Error across both sides. + // A chunk or the terminal end-of-stream marker. Chunks and the marker share one + // FIFO so the consumer can never observe end-of-stream while chunks remain buffered. + private readonly struct Item + { + public readonly TContent Chunk; + public readonly bool IsTerminal; + public readonly StreamError Error; + + private Item(TContent chunk, bool isTerminal, StreamError error) + { + Chunk = chunk; + IsTerminal = isTerminal; + Error = error; + } + + public static Item ForChunk(TContent chunk) => new Item(chunk, false, null); + public static Item ForTerminal(StreamError error) => new Item(default, true, error); + } + + // Items buffered beyond the one the consumer is currently positioned on. + private readonly Queue _queue = new(); + private Item _current; + private bool _hasCurrent; + + // Chunk/EOS events arrive on the FFI thread; Reset() and the LatestChunk getter run on + // the main-thread coroutine. _gate serializes mutations of the queue, the current item, + // IsCurrentReadDone, IsEos, and Error across both sides. private readonly object _gate = new(); /// @@ -99,10 +124,11 @@ public abstract class ReadIncrementalInstructionBase : StreamYieldInst /// /// The chunk from the most recent completed read. Throws the captured - /// if the last read errored. Internal so the optional - /// UniTask async-enumerable adapter (which has InternalsVisibleTo access) can read - /// it generically; the typed Bytes/Text accessors on the concrete - /// readers delegate here. + /// if the stream ended with an error. Returns the default + /// value once positioned on a normal end-of-stream marker (there is no chunk to read). + /// Internal so the optional UniTask async-enumerable adapter (which has + /// InternalsVisibleTo access) can read it generically; the typed Bytes/Text + /// accessors on the concrete readers delegate here. /// internal TContent LatestChunk { @@ -111,7 +137,8 @@ internal TContent LatestChunk lock (_gate) { if (Error != null) throw Error; - return _latestChunk; + if (!_hasCurrent) return default; + return _current.Chunk; } } } @@ -127,33 +154,37 @@ protected void OnChunk(TContent content) { lock (_gate) { - if (IsCurrentReadDone) + if (_hasCurrent) { - // Consumer hasn't yielded since the last chunk; buffer until Reset(). - _pendingChunks.Enqueue(content); + // Consumer hasn't advanced onto the current item yet; buffer in order. + _queue.Enqueue(Item.ForChunk(content)); } else { - _latestChunk = content; - IsCurrentReadDone = true; + Advance(Item.ForChunk(content)); } } } public override void Reset() { - // base.Reset() must run under the same lock as OnChunk, otherwise the - // window between IsCurrentReadDone=false (from base) and the dequeue - // below lets a producer race in, write _latestChunk, and have its - // chunk immediately overwritten by the dequeue. That race lost ~4% of - // chunks under stress before this fix. + // base.Reset() must run under the same lock as OnChunk/OnEos, otherwise the + // window between IsCurrentReadDone=false (from base) and the dequeue below lets + // a producer race in, position the current item, and have it immediately + // overwritten by the dequeue. base.Reset() also throws when the consumer tries to + // advance past the terminal marker (IsEos), which is the intended guard. lock (_gate) { base.Reset(); - if (_pendingChunks.Count > 0) + if (_queue.Count > 0) + { + Advance(_queue.Dequeue()); + } + else { - _latestChunk = _pendingChunks.Dequeue(); - IsCurrentReadDone = true; + // Nothing buffered: park until the next chunk or the terminal marker + // arrives. keepWaiting stays true until then. + _hasCurrent = false; } } } @@ -162,16 +193,38 @@ protected void OnEos(Proto.StreamError protoError) { lock (_gate) { - // Assign Error before flipping IsEos. The IsEos setter fires the awaiter - // continuation, which inspects IsError/Error on resume; when completion runs - // inline on the main thread, setting IsEos first would let the continuation - // observe IsError == false and silently swallow the stream error. - if (protoError != null) + var error = protoError != null ? new StreamError(protoError) : null; + if (_hasCurrent) { - Error = new StreamError(protoError); + // Ordered after every buffered chunk; the consumer drains them first. + _queue.Enqueue(Item.ForTerminal(error)); } + else + { + Advance(Item.ForTerminal(error)); + } + } + } + + // Positions the consumer on the given item and flips the matching completion flag. + // Caller must hold _gate. + private void Advance(Item item) + { + _current = item; + _hasCurrent = true; + if (item.IsTerminal) + { + // Assign Error before flipping IsEos. The IsEos setter fires the awaiter + // continuation, which inspects IsError/Error on resume; setting IsEos first + // would let the continuation observe IsError == false and silently swallow + // the stream error. + if (item.Error != null) Error = item.Error; IsEos = true; } + else + { + IsCurrentReadDone = true; + } } } diff --git a/Runtime/Scripts/DataStreams/TextDataStream.cs b/Runtime/Scripts/DataStreams/TextDataStream.cs index f6181a2c..bddcbb77 100644 --- a/Runtime/Scripts/DataStreams/TextDataStream.cs +++ b/Runtime/Scripts/DataStreams/TextDataStream.cs @@ -136,8 +136,9 @@ public ReadIncrementalInstruction ReadIncremental() /// YieldInstruction for . /// /// - /// Usage: while is false (i.e. the stream has not ended), - /// call , yield the instruction, and then access . + /// Usage: loop yielding the instruction; when it resumes, break if + /// is true (the stream has ended and every chunk has been drained), otherwise read + /// and call to advance to the next chunk. /// public sealed class ReadIncrementalInstruction : ReadIncrementalInstructionBase { diff --git a/Runtime/Scripts/UniTask/StreamReaderUniTaskExtensions.cs b/Runtime/Scripts/UniTask/StreamReaderUniTaskExtensions.cs index 45ea1284..5eb4f144 100644 --- a/Runtime/Scripts/UniTask/StreamReaderUniTaskExtensions.cs +++ b/Runtime/Scripts/UniTask/StreamReaderUniTaskExtensions.cs @@ -18,17 +18,13 @@ public static class StreamReaderUniTaskExtensions /// (string). /// /// - /// Iteration ends when the stream reaches end-of-stream. If the stream ends with an - /// error, the enumerable throws that (idiomatic for - /// await foreach; this is the one place the UniTask surface throws rather than - /// exposing IsError). Cancellation (via the token or the enumerator) surfaces as - /// with abandon-awaiter semantics — the - /// underlying FFI read is not cancelled on the wire. - /// - /// The current chunk is delivered on the iteration where end-of-stream is also observed, - /// then iteration stops. Chunks buffered beyond the current one when - /// end-of-stream arrives are not drainable, because the reader disallows Reset() - /// past end-of-stream. + /// Iteration ends when the stream reaches end-of-stream. Every buffered chunk is + /// drained before iteration stops, since end-of-stream is an ordered item at the tail + /// of the chunk queue. If the stream ends with an error, the enumerable throws that + /// (idiomatic for await foreach; this is the one place + /// the UniTask surface throws rather than exposing IsError). Cancellation (via + /// the token or the enumerator) surfaces as + /// with abandon-awaiter semantics — the underlying FFI read is not cancelled on the wire. /// public static IUniTaskAsyncEnumerable AsAsyncEnumerable( this ReadIncrementalInstructionBase instruction, @@ -44,30 +40,20 @@ public static IUniTaskAsyncEnumerable AsAsyncEnumerable( while (true) { - // Completes when a chunk is ready (IsCurrentReadDone) or the stream ends (IsEos). + // Completes when a chunk is ready (IsCurrentReadDone) or the consumer has + // advanced onto the terminal marker (IsEos) — the two are mutually exclusive. await instruction.AsUniTask(ct); if (instruction.IsCurrentReadDone) { var chunk = instruction.LatestChunk; await writer.YieldAsync(chunk); - - // Re-check IsEos AFTER yielding: end-of-stream may have arrived while - // the consumer was suspended. Reset() throws once IsEos is set, so this - // re-check (not a value captured before the yield) is what keeps the - // loop safe — mirroring the coroutine consumer's "if (IsEos) break; - // else Reset()" ordering. - if (instruction.IsEos) - { - if (instruction.IsError) throw instruction.Error; - return; - } - - instruction.Reset(); + instruction.Reset(); // advance to the next chunk or the terminal marker continue; } - // Not IsCurrentReadDone => end-of-stream with nothing left to read. + // Positioned on the terminal marker: the queue is fully drained. Surface an + // error close by throwing, otherwise end iteration cleanly. if (instruction.IsError) throw instruction.Error; return; } diff --git a/Samples~/Agents/Assets/Runtime/Agent/TranscriptionReader.cs b/Samples~/Agents/Assets/Runtime/Agent/TranscriptionReader.cs index aaa77186..e8e104c0 100644 --- a/Samples~/Agents/Assets/Runtime/Agent/TranscriptionReader.cs +++ b/Samples~/Agents/Assets/Runtime/Agent/TranscriptionReader.cs @@ -69,24 +69,18 @@ IEnumerator PumpNpc(TextStreamReader reader, string identity) var append = _beginReply(Speaker.Npc); var filter = new AnnotationFilter(); var r = reader.ReadIncremental(); - string lastSeen = null; while (true) { yield return r; + if (r.IsEos) break; - // EOS ticks don't refresh Text — only forward when the reference actually changed. var chunk = r.Text; - if (!ReferenceEquals(chunk, lastSeen)) + if (!string.IsNullOrEmpty(chunk)) { - if (!string.IsNullOrEmpty(chunk)) - { - var filtered = filter.Filter(chunk); - if (!string.IsNullOrEmpty(filtered)) append(filtered); - } - lastSeen = chunk; + var filtered = filter.Filter(chunk); + if (!string.IsNullOrEmpty(filtered)) append(filtered); } - if (r.IsEos) break; r.Reset(); } } From b12adc98567dcf22e40ad67015a62cb415cb2150 Mon Sep 17 00:00:00 2001 From: Max Heimbrock <43608204+MaxHeimbrock@users.noreply.github.com> Date: Mon, 20 Jul 2026 13:49:38 +0200 Subject: [PATCH 3/4] Simplification of code --- Runtime/Scripts/DataStreams/DataStream.cs | 84 ++++++++--------------- 1 file changed, 29 insertions(+), 55 deletions(-) diff --git a/Runtime/Scripts/DataStreams/DataStream.cs b/Runtime/Scripts/DataStreams/DataStream.cs index dd88b469..13bf6685 100644 --- a/Runtime/Scripts/DataStreams/DataStream.cs +++ b/Runtime/Scripts/DataStreams/DataStream.cs @@ -85,6 +85,8 @@ public abstract class ReadIncrementalInstructionBase : StreamYieldInst // A chunk or the terminal end-of-stream marker. Chunks and the marker share one // FIFO so the consumer can never observe end-of-stream while chunks remain buffered. + // The head of the queue is the item the consumer is currently positioned on; Reset() + // dequeues it to advance to the next. private readonly struct Item { public readonly TContent Chunk; @@ -102,14 +104,11 @@ private Item(TContent chunk, bool isTerminal, StreamError error) public static Item ForTerminal(StreamError error) => new Item(default, true, error); } - // Items buffered beyond the one the consumer is currently positioned on. private readonly Queue _queue = new(); - private Item _current; - private bool _hasCurrent; // Chunk/EOS events arrive on the FFI thread; Reset() and the LatestChunk getter run on - // the main-thread coroutine. _gate serializes mutations of the queue, the current item, - // IsCurrentReadDone, IsEos, and Error across both sides. + // the main-thread coroutine. _gate serializes mutations of the queue, IsCurrentReadDone, + // IsEos, and Error across both sides. private readonly object _gate = new(); /// @@ -137,8 +136,7 @@ internal TContent LatestChunk lock (_gate) { if (Error != null) throw Error; - if (!_hasCurrent) return default; - return _current.Chunk; + return _queue.Count > 0 ? _queue.Peek().Chunk : default; } } } @@ -150,75 +148,51 @@ protected ReadIncrementalInstructionBase(FfiHandle readerHandle) protected bool MatchesHandle(ulong eventHandle) => eventHandle == _handleValue; - protected void OnChunk(TContent content) + protected void OnChunk(TContent content) => Enqueue(Item.ForChunk(content)); + + protected void OnEos(Proto.StreamError protoError) => + Enqueue(Item.ForTerminal(protoError != null ? new StreamError(protoError) : null)); + + private void Enqueue(Item item) { lock (_gate) { - if (_hasCurrent) - { - // Consumer hasn't advanced onto the current item yet; buffer in order. - _queue.Enqueue(Item.ForChunk(content)); - } - else - { - Advance(Item.ForChunk(content)); - } + // When the queue was empty this item becomes the head, so publish it now; + // otherwise it stays buffered in order behind the current head. + bool wasEmpty = _queue.Count == 0; + _queue.Enqueue(item); + if (wasEmpty) SettleHead(); } } public override void Reset() { - // base.Reset() must run under the same lock as OnChunk/OnEos, otherwise the - // window between IsCurrentReadDone=false (from base) and the dequeue below lets - // a producer race in, position the current item, and have it immediately - // overwritten by the dequeue. base.Reset() also throws when the consumer tries to - // advance past the terminal marker (IsEos), which is the intended guard. + // base.Reset() must run under the same lock as OnChunk/OnEos so a producer can't + // race between the flag clear and the dequeue below. It also throws when the + // consumer tries to advance past the terminal marker (IsEos), the intended guard. lock (_gate) { base.Reset(); - if (_queue.Count > 0) - { - Advance(_queue.Dequeue()); - } - else - { - // Nothing buffered: park until the next chunk or the terminal marker - // arrives. keepWaiting stays true until then. - _hasCurrent = false; - } - } - } - - protected void OnEos(Proto.StreamError protoError) - { - lock (_gate) - { - var error = protoError != null ? new StreamError(protoError) : null; - if (_hasCurrent) - { - // Ordered after every buffered chunk; the consumer drains them first. - _queue.Enqueue(Item.ForTerminal(error)); - } - else - { - Advance(Item.ForTerminal(error)); - } + if (_queue.Count > 0) _queue.Dequeue(); // drop the item just consumed + SettleHead(); } } - // Positions the consumer on the given item and flips the matching completion flag. + // Reflects the queue head in the base completion flags: a chunk becomes readable + // (IsCurrentReadDone), the terminal marker ends the stream (IsEos). An empty queue + // parks the consumer (keepWaiting stays true) until the next item arrives. // Caller must hold _gate. - private void Advance(Item item) + private void SettleHead() { - _current = item; - _hasCurrent = true; - if (item.IsTerminal) + if (_queue.Count == 0) return; + var head = _queue.Peek(); + if (head.IsTerminal) { // Assign Error before flipping IsEos. The IsEos setter fires the awaiter // continuation, which inspects IsError/Error on resume; setting IsEos first // would let the continuation observe IsError == false and silently swallow // the stream error. - if (item.Error != null) Error = item.Error; + if (head.Error != null) Error = head.Error; IsEos = true; } else From 2c6e7893a36570946d4cd3c627a89caea028b2bd Mon Sep 17 00:00:00 2001 From: Max Heimbrock <43608204+MaxHeimbrock@users.noreply.github.com> Date: Mon, 20 Jul 2026 16:11:12 +0200 Subject: [PATCH 4/4] Terminal end of stream marker renamed to isEndOfStream --- Runtime/Scripts/DataStreams/DataStream.cs | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/Runtime/Scripts/DataStreams/DataStream.cs b/Runtime/Scripts/DataStreams/DataStream.cs index 13bf6685..fd5f7eab 100644 --- a/Runtime/Scripts/DataStreams/DataStream.cs +++ b/Runtime/Scripts/DataStreams/DataStream.cs @@ -73,7 +73,7 @@ internal StreamError(Proto.StreamError proto) : base(proto.Description) { } /// /// Shared state and helpers for incremental stream reader yield instructions. /// Models the incoming stream as a single ordered queue whose items are either a - /// chunk or a terminal end-of-stream marker (carrying an optional ). + /// chunk or a the end-of-stream marker (carrying an optional ). /// The consumer advances through the queue one item at a time via ; /// end-of-stream is only observed once every buffered chunk has been drained, mirroring /// the JS and Python SDKs. Subclasses own the typed event subscription and convert raw @@ -83,25 +83,25 @@ public abstract class ReadIncrementalInstructionBase : StreamYieldInst { private readonly ulong _handleValue; - // A chunk or the terminal end-of-stream marker. Chunks and the marker share one + // A chunk or the end-of-stream marker. Chunks and the marker share one // FIFO so the consumer can never observe end-of-stream while chunks remain buffered. // The head of the queue is the item the consumer is currently positioned on; Reset() // dequeues it to advance to the next. private readonly struct Item { public readonly TContent Chunk; - public readonly bool IsTerminal; + public readonly bool IsEos; public readonly StreamError Error; - private Item(TContent chunk, bool isTerminal, StreamError error) + private Item(TContent chunk, bool isEos, StreamError error) { Chunk = chunk; - IsTerminal = isTerminal; + IsEos = isEos; Error = error; } public static Item ForChunk(TContent chunk) => new Item(chunk, false, null); - public static Item ForTerminal(StreamError error) => new Item(default, true, error); + public static Item ForEos(StreamError error) => new Item(default, true, error); } private readonly Queue _queue = new(); @@ -151,7 +151,7 @@ protected ReadIncrementalInstructionBase(FfiHandle readerHandle) protected void OnChunk(TContent content) => Enqueue(Item.ForChunk(content)); protected void OnEos(Proto.StreamError protoError) => - Enqueue(Item.ForTerminal(protoError != null ? new StreamError(protoError) : null)); + Enqueue(Item.ForEos(protoError != null ? new StreamError(protoError) : null)); private void Enqueue(Item item) { @@ -169,7 +169,7 @@ public override void Reset() { // base.Reset() must run under the same lock as OnChunk/OnEos so a producer can't // race between the flag clear and the dequeue below. It also throws when the - // consumer tries to advance past the terminal marker (IsEos), the intended guard. + // consumer tries to advance past the Eos marker (IsEos), the intended guard. lock (_gate) { base.Reset(); @@ -179,14 +179,14 @@ public override void Reset() } // Reflects the queue head in the base completion flags: a chunk becomes readable - // (IsCurrentReadDone), the terminal marker ends the stream (IsEos). An empty queue + // (IsCurrentReadDone), the Eos marker (IsEos) ends the stream. An empty queue // parks the consumer (keepWaiting stays true) until the next item arrives. // Caller must hold _gate. private void SettleHead() { if (_queue.Count == 0) return; var head = _queue.Peek(); - if (head.IsTerminal) + if (head.IsEos) { // Assign Error before flipping IsEos. The IsEos setter fires the awaiter // continuation, which inspects IsError/Error on resume; setting IsEos first