fix: streaming robustness and refactor

e2e/react-start/streaming-ssr/tests/fast-serial.spec.ts

@@ -16,6 +16,21 @@ test.describe('Fast serialization (serialization completes before render)', () =
     expect(responseHtml).toContain('$_TSR.router')
     expect(responseHtml).toContain('$_TSR.e()')
     expect(responseHtml).toContain('$tsr-stream-barrier')
+
+    // Router scripts and end marker must be injected before closing tags.
+    const bodyCloseIndex = responseHtml.lastIndexOf('</body>')
+    const htmlCloseIndex = responseHtml.lastIndexOf('</html>')
+    const barrierIndex = responseHtml.indexOf('$tsr-stream-barrier')
+    const endMarkerIndex = responseHtml.indexOf('$_TSR.e()')
+
+    expect(bodyCloseIndex).toBeGreaterThan(-1)
+    expect(htmlCloseIndex).toBeGreaterThan(-1)
+    expect(barrierIndex).toBeGreaterThan(-1)
+    expect(endMarkerIndex).toBeGreaterThan(-1)
+
+    expect(barrierIndex).toBeLessThan(bodyCloseIndex)
+    expect(endMarkerIndex).toBeLessThan(bodyCloseIndex)
+    expect(endMarkerIndex).toBeLessThan(htmlCloseIndex)
   })
 
   test('all data is available immediately', async ({ page }) => {

e2e/react-start/streaming-ssr/tests/sync-only.spec.ts

@@ -68,6 +68,21 @@ test('Sync-only route has bootstrap scripts in initial HTML', async ({
   // SSR should include the barrier script tag in the HTML (rendered by <Scripts />)
   // This is the critical marker transformStreamWithRouter can scan for.
   expect(responseHtml).toContain('$tsr-stream-barrier')
+
+  // Router scripts and end marker must be injected before closing tags.
+  const bodyCloseIndex = responseHtml.lastIndexOf('</body>')
+  const htmlCloseIndex = responseHtml.lastIndexOf('</html>')
+  const barrierIndex = responseHtml.indexOf('$tsr-stream-barrier')
+  const endMarkerIndex = responseHtml.indexOf('$_TSR.e()')
+
+  expect(bodyCloseIndex).toBeGreaterThan(-1)
+  expect(htmlCloseIndex).toBeGreaterThan(-1)
+  expect(barrierIndex).toBeGreaterThan(-1)
+  expect(endMarkerIndex).toBeGreaterThan(-1)
+
+  expect(barrierIndex).toBeLessThan(bodyCloseIndex)
+  expect(endMarkerIndex).toBeLessThan(bodyCloseIndex)
+  expect(endMarkerIndex).toBeLessThan(htmlCloseIndex)
 })
 
 test('Navigating to sync-only from home page', async ({ page }) => {

packages/router-core/package.json

@@ -31,6 +31,11 @@
     "test:build": "publint --strict && attw --ignore-rules no-resolution --pack .",
     "test:unit": "vitest",
     "test:unit:dev": "pnpm run test:unit --watch",
+    "bench:transform-stream": "vitest bench tests/transformStreamWithRouter.bench.ts",
+    "bench:transform-stream:prof": "pnpm dlx @platformatic/flame run --md-format=detailed ../../node_modules/vitest/vitest.mjs bench tests/transformStreamWithRouter.bench.ts",
+    "profile:transform-stream": "pnpm build && pnpm run profile:transform-stream:run:prof",
+    "profile:transform-stream:run": "node ./tests/transformStreamWithRouter.profile.mjs",
+    "profile:transform-stream:run:prof": "pnpm dlx @platformatic/flame run --delay=none --md-format=detailed ./tests/transformStreamWithRouter.profile.mjs",
     "build": "vite build"
   },
   "type": "module",

packages/router-core/src/ssr/ssr-server.ts

@@ -91,13 +91,8 @@ class ScriptBuffer {
   liftBarrier() {
     if (this._scriptBarrierLifted || this._cleanedUp) return
     this._scriptBarrierLifted = true
-    if (this._queue.length > 0 && !this._pendingMicrotask) {
-      this._pendingMicrotask = true
-      queueMicrotask(() => {
-        this._pendingMicrotask = false
-        this.injectBufferedScripts()
-      })
-    }
+    this._pendingMicrotask = false
+    this.injectBufferedScripts()
   }
 
   /**
@@ -274,15 +269,24 @@ export function attachRouterServerSsrUtils({
         : defaultSerovalPlugins
 
       const signalSerializationComplete = () => {
+        if (_serializationFinished) return
         _serializationFinished = true
+
+        const listeners = serializationFinishedListeners.slice()
+
+        for (const listener of listeners) {
+          try {
+            listener()
+          } catch (err) {
+            console.error('Serialization listener error:', err)
+          }
+        }
         try {
-          serializationFinishedListeners.forEach((l) => l())
           router.emit({ type: 'onSerializationFinished' })
         } catch (err) {
-          console.error('Serialization listener error:', err)
+          console.error('Error emitting onSerializationFinished:', err)
         } finally {
           serializationFinishedListeners.length = 0
-          renderFinishedListeners.length = 0
         }
       }
 
@@ -320,16 +324,19 @@ export function attachRouterServerSsrUtils({
     onSerializationFinished: (listener) =>
       serializationFinishedListeners.push(listener),
     setRenderFinished: () => {
-      // Wrap in try-catch to ensure scriptBuffer.liftBarrier() is always called
-      try {
-        renderFinishedListeners.forEach((l) => l())
-      } catch (err) {
-        console.error('Error in render finished listener:', err)
-      } finally {
-        // Clear listeners after calling them to prevent memory leaks
-        renderFinishedListeners.length = 0
+      const listeners = renderFinishedListeners.slice()
+      renderFinishedListeners.length = 0
+
+      for (const listener of listeners) {
+        try {
+          listener()
+        } catch (err) {
+          console.error('Error in render finished listener:', err)
+        }
       }
+
       scriptBuffer.liftBarrier()
+      scriptBuffer.flush()
     },
     takeBufferedScripts() {
       const scripts = scriptBuffer.takeAll()
@@ -396,7 +403,7 @@ export function getOrigin(request: Request) {
 // chromium treats search params differently than paths, i.e. "|" is not encoded in search params.
 export function getNormalizedURL(url: string | URL, base?: string | URL) {
   // ensure backslashes are encoded correctly in the URL
-  if (typeof url === 'string') url = url.replace('\\', '%5C')
+  if (typeof url === 'string') url = url.replace(/\\/g, '%5C')
 
   const rawUrl = new URL(url, base)
   const { path: decodedPathname, handledProtocolRelativeURL } = decodePath(