fix(cli): Identify and fix a test deadlock (#23411)

If a worker tried to flush large amounts of data right as the test was
ending, it could cause the flush sync marker to get lost.

1 files changed, 26 insertions, 2 deletions

diff --git a/cli/tools/test/channel.rs b/cli/tools/test/channel.rs
index aad3f926e..780a17de6 100644
--- a/cli/tools/test/channel.rs
+++ b/cli/tools/test/channel.rs
@@ -10,6 +10,7 @@ use deno_runtime::deno_io::pipe;
 use deno_runtime::deno_io::AsyncPipeRead;
 use deno_runtime::deno_io::PipeRead;
 use deno_runtime::deno_io::PipeWrite;
+use memmem::Searcher;
 use std::fmt::Display;
 use std::future::Future;
 use std::io::Write;
@@ -30,6 +31,7 @@ use tokio::sync::mpsc::WeakUnboundedSender;
 /// 8-byte sync marker that is unlikely to appear in normal output. Equivalent
 /// to the string `"\u{200B}\0\u{200B}\0"`.
 const SYNC_MARKER: &[u8; 8] = &[226, 128, 139, 0, 226, 128, 139, 0];
+const HALF_SYNC_MARKER: &[u8; 4] = &[226, 128, 139, 0];
 
 const BUFFER_SIZE: usize = 4096;
 
@@ -202,8 +204,30 @@ impl TestStream {
         }
         Ok(read) => {
           flush.extend(&buffer[0..read]);
-          if flush.ends_with(SYNC_MARKER) {
-            flush.truncate(flush.len() - SYNC_MARKER.len());
+
+          // "ends_with" is cheaper, so check that first
+          if flush.ends_with(HALF_SYNC_MARKER) {
+            // We might have read the full sync marker.
+            if flush.ends_with(SYNC_MARKER) {
+              flush.truncate(flush.len() - SYNC_MARKER.len());
+            } else {
+              flush.truncate(flush.len() - HALF_SYNC_MARKER.len());
+            }
+            // Try to send our flushed buffer. If the channel is closed, this stream will
+            // be marked as not alive.
+            _ = self.send(flush);
+            return;
+          }
+
+          // If we don't end with the marker, then we need to search the bytes we read plus four bytes
+          // from before. There's still a possibility that the marker could be split because of a pipe
+          // buffer that fills up, forcing the flush to be written across two writes and interleaving
+          // data between, but that's a risk we take with this sync marker approach.
+          let searcher = memmem::TwoWaySearcher::new(HALF_SYNC_MARKER);
+          let start =
+            (flush.len() - read).saturating_sub(HALF_SYNC_MARKER.len());
+          if let Some(offset) = searcher.search_in(&flush[start..]) {
+            flush.truncate(offset);
             // Try to send our flushed buffer. If the channel is closed, this stream will
             // be marked as not alive.
             _ = self.send(flush);