diff --git a/packages/client/src/adapters/standard/rpc-link-codec.ts b/packages/client/src/adapters/standard/rpc-link-codec.ts
index 508cf792a..9b9ef66b3 100644
--- a/packages/client/src/adapters/standard/rpc-link-codec.ts
+++ b/packages/client/src/adapters/standard/rpc-link-codec.ts
@@ -1,5 +1,11 @@
 import type { Promisable, Public, Value } from '@orpc/shared'
-import { isAsyncIteratorObject, pathToHttpPath, stringifyJSON, value } from '@orpc/shared'
+import {
+  isAsyncIteratorObject,
+  mergeHttpPath,
+  pathToHttpPath,
+  stringifyJSON,
+  value,
+} from '@orpc/shared'
 import type {
   StandardHeaders,
   StandardLazyResponse,
@@ -74,8 +80,6 @@ export interface RPCLinkCodecOptions<T extends ClientContext> {
   serializer?: Public<RPCSerializer> | undefined
 }
 
-const END_SLASH_REGEX = /\/$/
-
 export class RPCLinkCodec<T extends ClientContext> implements StandardLinkCodec<T> {
   private readonly baseUrl: Exclude<RPCLinkCodecOptions<T>['url'], undefined>
   private readonly maxUrlLength: Exclude<RPCLinkCodecOptions<T>['maxUrlLength'], undefined>
@@ -107,8 +111,7 @@ export class RPCLinkCodec<T extends ClientContext> implements StandardLinkCodec<
     const baseUrl = await value(this.baseUrl, options, path, input)
 
     const [pathname, search, hash] = parseStandardUrl(baseUrl)
-    const newPathname =
-      `${pathname.replace(END_SLASH_REGEX, '')}${pathToHttpPath(path)}` as StandardUrl
+    const newPathname: StandardUrl = mergeHttpPath(pathname, pathToHttpPath(path))
     const serialized = this.serializer.serialize(input)
 
     if (
diff --git a/packages/client/src/plugins/dedupe.ts b/packages/client/src/plugins/dedupe.ts
index 777da10c5..fba88965d 100644
--- a/packages/client/src/plugins/dedupe.ts
+++ b/packages/client/src/plugins/dedupe.ts
@@ -9,6 +9,7 @@ import {
   defer,
   disposableAllAbortSignal,
   isAsyncIteratorObject,
+  listenAbort,
   replicateAsyncIterator,
   replicateReadableStream,
   runWithSignal,
@@ -318,13 +319,7 @@ function closeAsyncIteratorOnAbort<T, TReturn, TNext>(
     return replica
   }
 
-  const abort = () => replica.return().catch(() => {})
-
-  if (signal.aborted) {
-    void abort()
-  } else {
-    signal.addEventListener('abort', abort, { once: true })
-  }
+  const removeAbortListener = listenAbort(signal, () => void replica.return().catch(() => {}))
 
   return wrapAsyncIterator(replica, {
     mapResult(result) {
@@ -332,7 +327,7 @@ function closeAsyncIteratorOnAbort<T, TReturn, TNext>(
       throwIfAborted(signal)
       return result
     },
-    onFinish: () => signal.removeEventListener('abort', abort),
+    onFinish: removeAbortListener,
   })
 }
 
@@ -349,13 +344,9 @@ function closeReadableStreamOnAbort<T>(
   }
 
   const reader = replica.getReader()
-  const abort = () => reader.cancel(signal.reason).catch(() => {})
-
-  if (signal.aborted) {
-    void abort()
-  } else {
-    signal.addEventListener('abort', abort, { once: true })
-  }
+  const removeAbortListener = listenAbort(signal, (reason) => {
+    reader.cancel(reason).catch(() => {})
+  })
 
   return new ReadableStream<T>(
     {
@@ -366,18 +357,18 @@ function closeReadableStreamOnAbort<T>(
           throwIfAborted(signal)
 
           if (result.done) {
-            signal.removeEventListener('abort', abort)
+            removeAbortListener()
             controller.close()
           } else {
             controller.enqueue(result.value)
           }
         } catch (error) {
-          signal.removeEventListener('abort', abort)
+          removeAbortListener()
           throw error
         }
       },
       async cancel(reason) {
-        signal.removeEventListener('abort', abort)
+        removeAbortListener()
         await reader.cancel(reason)
       },
     },
diff --git a/packages/client/src/plugins/request-compression.ts b/packages/client/src/plugins/request-compression.ts
index 49943adee..10f97001f 100644
--- a/packages/client/src/plugins/request-compression.ts
+++ b/packages/client/src/plugins/request-compression.ts
@@ -1,4 +1,11 @@
-import { isAsyncIteratorObject, isCompressibleContentType, toArray } from '@orpc/shared'
+import type { CompressionEncoding } from '@orpc/shared'
+import {
+  estimateUtf8ByteLength,
+  estimateCompressibleFormDataSize,
+  isAsyncIteratorObject,
+  isCompressibleContentType,
+  toArray,
+} from '@orpc/shared'
 import type { StandardBodyHint, StandardHeaders } from '@standard-server/core'
 import { flattenStandardHeader, generateContentDisposition } from '@standard-server/core'
 import { toFetchBody } from '@standard-server/fetch'
@@ -11,17 +18,13 @@ import type {
 } from '../adapters/standard/index.ts'
 import type { ClientContext } from '../types.ts'
 
-// Rough UTF-8 estimate. Mostly ASCII text stays close to 1 byte/char;
-// occasional multi-byte characters increase the average.
-const AVG_BYTES_PER_CHAR = 1.2
-
 export interface RequestCompressionLinkPluginOptions<T extends ClientContext> {
   /**
    * The compression scheme to use for request compression.
    *
    * @default 'gzip'
    */
-  encoding?: 'gzip' | 'deflate' | 'deflate-raw' | undefined
+  encoding?: CompressionEncoding | undefined
 
   /**
    * The minimum request size in bytes required to trigger compression.
@@ -151,32 +154,9 @@ export class RequestCompressionLinkPlugin<
           })
         }
       } else if (request.body instanceof FormData) {
-        const PART_OVERHEAD = 64 // approx bytes for boundary + Content-Disposition/Content-Type headers per part
-
-        let contentLength = 0
-        for (const [key, value] of request.body) {
-          contentLength += PART_OVERHEAD + key.length
-
-          if (value instanceof Blob) {
-            if (!Number.isFinite(value.size)) {
-              // Bun-s3 can use NaN for size
-              if (!this.isCompressibleContentType(value.type, interceptorOptions)) {
-                // Unknown non-compressible part size makes the estimate unreliable
-                contentLength = -Infinity
-                break
-              }
-
-              // Unknown size for compressible content - still apply compression
-              contentLength = Infinity
-            } else {
-              contentLength += this.isCompressibleContentType(value.type, interceptorOptions)
-                ? value.size
-                : -value.size
-            }
-          } else {
-            contentLength += value.length * AVG_BYTES_PER_CHAR
-          }
-        }
+        const contentLength = estimateCompressibleFormDataSize(request.body, (contentType) =>
+          this.isCompressibleContentType(contentType, interceptorOptions),
+        )
 
         if (contentLength >= this.threshold) {
           const res = new Response(request.body)
@@ -199,7 +179,7 @@ export class RequestCompressionLinkPlugin<
         }
       } else if (request.body instanceof URLSearchParams) {
         const string = request.body.toString()
-        if (string.length * AVG_BYTES_PER_CHAR >= this.threshold) {
+        if (estimateUtf8ByteLength(string) >= this.threshold) {
           const compressedStream = new Blob([string])
             .stream()
             .pipeThrough(new CompressionStream(this.encoding))
@@ -225,7 +205,7 @@ export class RequestCompressionLinkPlugin<
           string,
           StandardHeaders,
         ]
-        if (string.length * AVG_BYTES_PER_CHAR >= this.threshold) {
+        if (estimateUtf8ByteLength(string) >= this.threshold) {
           const compressedStream = new Blob([string])
             .stream()
             .pipeThrough(new CompressionStream(this.encoding))
diff --git a/packages/client/src/plugins/response-compression.test.ts b/packages/client/src/plugins/response-compression.test.ts
index 533396577..da0a65846 100644
--- a/packages/client/src/plugins/response-compression.test.ts
+++ b/packages/client/src/plugins/response-compression.test.ts
@@ -115,6 +115,22 @@ describe('responseCompressionLinkPlugin', () => {
     await expect(link.call(['test'], undefined, { context: {} })).resolves.toEqual('OK')
   })
 
+  it('ignores empty content-encoding list elements', async () => {
+    const compressed = await compressAsync(JSON.stringify({ json: 'OK' }), 'gzip')
+
+    const { link } = createLink({
+      fetchImpl: async () =>
+        new Response(compressed, {
+          headers: {
+            'content-type': 'application/json',
+            'content-encoding': 'gzip, ',
+          },
+        }),
+    })
+
+    await expect(link.call(['test'], undefined, { context: {} })).resolves.toEqual('OK')
+  })
+
   it('decompresses response body when 5 content-encodings are applied', async () => {
     let body: Uint8Array<ArrayBuffer> = await compressAsync(JSON.stringify({ json: 'OK' }), 'gzip')
     for (let i = 1; i < 5; i++) {
diff --git a/packages/client/src/plugins/response-compression.ts b/packages/client/src/plugins/response-compression.ts
index 2caf1869b..fbd19fbed 100644
--- a/packages/client/src/plugins/response-compression.ts
+++ b/packages/client/src/plugins/response-compression.ts
@@ -1,4 +1,5 @@
-import { toArray } from '@orpc/shared'
+import type { CompressionEncoding } from '@orpc/shared'
+import { isCompressionEncoding, parseContentEncodings, toArray } from '@orpc/shared'
 import { flattenStandardHeader } from '@standard-server/core'
 import { toFetchHeaders, toStandardBody } from '@standard-server/fetch'
 
@@ -22,7 +23,7 @@ export interface ResponseCompressionLinkPluginOptions<_T extends ClientContext>
    *
    * @default ['gzip', 'deflate']
    */
-  encodings?: readonly ('gzip' | 'deflate' | 'deflate-raw')[] | undefined
+  encodings?: readonly CompressionEncoding[] | undefined
 }
 
 /**
@@ -73,7 +74,7 @@ export class ResponseCompressionLinkPlugin<
         flattenStandardHeader(response.headers['content-encoding']),
       )
 
-      if (encodings.length === 0 || !encodings.every(isSupportedEncoding)) {
+      if (encodings.length === 0 || !encodings.every(isCompressionEncoding)) {
         return response
       }
 
@@ -120,23 +121,3 @@ export class ResponseCompressionLinkPlugin<
     }
   }
 }
-
-const SUPPORTED_ENCODINGS = ['gzip', 'deflate', 'deflate-raw'] as const
-type SupportedEncoding = (typeof SUPPORTED_ENCODINGS)[number]
-
-function isSupportedEncoding(encoding: string): encoding is SupportedEncoding {
-  return (SUPPORTED_ENCODINGS as readonly string[]).includes(encoding)
-}
-
-/**
- * Parse Content-Encoding into ordered codings (order applied).
- *
- * @see https://www.rfc-editor.org/rfc/rfc9110.html#name-content-encoding
- */
-function parseContentEncodings(header: string | undefined): string[] {
-  if (header === undefined) {
-    return []
-  }
-
-  return header.split(',').map((part) => part.trim().toLowerCase())
-}
diff --git a/packages/client/src/rpc-json-serializer.test.ts b/packages/client/src/rpc-json-serializer.test.ts
index 9b306035a..17b2f7df0 100644
--- a/packages/client/src/rpc-json-serializer.test.ts
+++ b/packages/client/src/rpc-json-serializer.test.ts
@@ -291,6 +291,19 @@ describe('rpcJsonSerializer: custom handlers', () => {
     expect(roundTripThroughWire(serializer, { value: date })).toEqual({ value: date })
   })
 
+  it('serializes like the defaults when no custom handler is set', () => {
+    const serializer = new RPCJsonSerializer({
+      handlers: {
+        unused: undefined,
+      },
+    })
+
+    const date = new Date('2023-01-01')
+    const data = { value: date, list: [undefined], text: 'a' }
+    expect(serializer.serializeJson(data)).toEqual(new RPCJsonSerializer().serializeJson(data))
+    expect(roundTripThroughWire(serializer, data)).toEqual(data)
+  })
+
   it('supports disabling default handlers', () => {
     const serializer = new RPCJsonSerializer({
       handlers: {
diff --git a/packages/client/src/rpc-json-serializer.ts b/packages/client/src/rpc-json-serializer.ts
index e6f4ef5e4..47010775c 100644
--- a/packages/client/src/rpc-json-serializer.ts
+++ b/packages/client/src/rpc-json-serializer.ts
@@ -3,6 +3,7 @@ import {
   copyOnWrite,
   DEFAULT_MAX_BIGINT_DIGITS,
   isPlainObject,
+  isPlainObjectOrArray,
   NullProtoObj,
   parseBigIntString,
 } from '@orpc/shared'
@@ -50,14 +51,10 @@ function assertSerializedType(ok: boolean, type: string, expected: string): void
   }
 }
 
-function isWalkable(value: unknown): boolean {
-  return Array.isArray(value) || isPlainObject(value)
-}
-
 function walkInto(parent: any, key: Segment, original: unknown, restored?: Set<unknown>): any {
   const value = parent[key]
 
-  if (!isWalkable(value) || restored?.has(value)) {
+  if (!isPlainObjectOrArray(value) || restored?.has(value)) {
     throw invalidSerializedData(`segment "${key}" cannot be walked into.`)
   }
 
@@ -562,7 +559,7 @@ export class RPCJsonSerializer {
         const value = handler.deserialize(currentRef[preSegment])
         currentRef[preSegment] = value
 
-        if (isWalkable(value)) {
+        if (isPlainObjectOrArray(value)) {
           ;(restored ??= new Set()).add(value)
         }
       }
diff --git a/packages/client/src/rpc-serializer.ts b/packages/client/src/rpc-serializer.ts
index 54bcb00e0..55d16d6e2 100644
--- a/packages/client/src/rpc-serializer.ts
+++ b/packages/client/src/rpc-serializer.ts
@@ -1,4 +1,9 @@
-import { isAsyncIteratorObject, isPlainObject, stringifyJSON } from '@orpc/shared'
+import {
+  isAsyncIteratorObject,
+  isPlainObject,
+  NON_NEGATIVE_INTEGER_REGEX,
+  stringifyJSON,
+} from '@orpc/shared'
 import type { StandardBody } from '@standard-server/core'
 import { EventStreamErrorEvent } from '@standard-server/core'
 
@@ -137,7 +142,7 @@ export class RPCSerializer extends RPCJsonSerializer {
     for (const [key, value] of data) {
       if (key === 'data' && raw === undefined && typeof value === 'string') {
         raw = value
-      } else if (value instanceof Blob && BLOB_KEY_REGEX.test(key)) {
+      } else if (value instanceof Blob && NON_NEGATIVE_INTEGER_REGEX.test(key)) {
         blobs[Number(key)] = value
         hasBlobs = true
       } else {
@@ -156,8 +161,6 @@ export class RPCSerializer extends RPCJsonSerializer {
   }
 }
 
-const BLOB_KEY_REGEX = /^(?:0|[1-9]\d*)$/
-
 function invalidFormat(): TypeError {
   return new TypeError('Invalid RPC serialized data format.')
 }
diff --git a/packages/json-schema/src/coercer.test.ts b/packages/json-schema/src/coercer.test.ts
index 92fe2672a..8020176b7 100644
--- a/packages/json-schema/src/coercer.test.ts
+++ b/packages/json-schema/src/coercer.test.ts
@@ -191,6 +191,9 @@ describe('jsonSchemaCoercer', () => {
       expect(coerce(DATE_SCHEMA, '2020-02-30T06:15:00Z')).toBe('2020-02-30T06:15:00Z')
       expect(coerce(DATE_SCHEMA, '2020-02-30 06:15')).toBe('2020-02-30 06:15')
       expect(coerce(DATE_SCHEMA, '0050-02-29')).toBe('0050-02-29')
+      // century years are leap years only when divisible by 400
+      expect(coerce(DATE_SCHEMA, '1900-02-29')).toBe('1900-02-29')
+      expect(coerce(DATE_SCHEMA, '2000-02-29')).toEqual(new Date('2000-02-29'))
 
       // real days stay coerced, including a leap day and years below 100
       expect(coerce(DATE_SCHEMA, '2020-02-29')).toEqual(new Date('2020-02-29'))
diff --git a/packages/json-schema/src/coercer.ts b/packages/json-schema/src/coercer.ts
index 632b7b70f..48f1c84dc 100644
--- a/packages/json-schema/src/coercer.ts
+++ b/packages/json-schema/src/coercer.ts
@@ -1,8 +1,10 @@
 import {
   DEFAULT_MAX_BIGINT_DIGITS,
   getOwn,
-  isPlainObject,
+  INTEGER_REGEX,
   isNonPrimitive,
+  isPlainObject,
+  isPlainObjectOrArray,
   parseBigIntString,
   toArray,
   tryOrUndefined,
@@ -553,7 +555,7 @@ function countValues(input: unknown, limit: number): number {
   while (stack.length) {
     const value = stack.pop()
 
-    if ((Array.isArray(value) || isPlainObject(value)) && !walked.has(value)) {
+    if (isPlainObjectOrArray(value) && !walked.has(value)) {
       walked.add(value)
 
       const items = Array.isArray(value) ? value : Object.values(value)
@@ -591,9 +593,8 @@ function stringToNumber(value: string): number | string {
   return num
 }
 
-const INTEGER_PATTERN = /^-?(?:0|[1-9]\d*)$/
 function stringToInteger(value: string): number | string {
-  if (!INTEGER_PATTERN.test(value)) {
+  if (!INTEGER_REGEX.test(value)) {
     return value
   }
 
diff --git a/packages/mcp/src/registrar.ts b/packages/mcp/src/registrar.ts
index e7db6d202..c61a57342 100644
--- a/packages/mcp/src/registrar.ts
+++ b/packages/mcp/src/registrar.ts
@@ -52,7 +52,7 @@ import {
   encodeBase64,
   getOwn,
   isAsyncIteratorObject,
-  isPlainObject,
+  isPlainObjectOrArray,
   loadBytes,
   pathToHttpPath,
   resolveMaybeOptionalOptions,
@@ -538,7 +538,7 @@ function collectNestedBlobs(
 ): void {
   if (json instanceof Blob) {
     blobs.push([toResourceUri(segments, json), json])
-  } else if (Array.isArray(json) || isPlainObject(json)) {
+  } else if (isPlainObjectOrArray(json)) {
     for (const key of Object.keys(json)) {
       segments.push(key)
       collectNestedBlobs((json as Record<string, unknown>)[key], segments, blobs)
diff --git a/packages/msw/src/http-procedure-utils.ts b/packages/msw/src/http-procedure-utils.ts
index 681620819..99fc61bfe 100644
--- a/packages/msw/src/http-procedure-utils.ts
+++ b/packages/msw/src/http-procedure-utils.ts
@@ -11,7 +11,7 @@ import type { Context, ProcedureConfig, ProcedureHandler, Router } from '@orpc/s
 import { implement } from '@orpc/server'
 import type { FetchHandler, FetchHandlerHandleResult } from '@orpc/server/fetch'
 import type { Promisable, Value } from '@orpc/shared'
-import { value } from '@orpc/shared'
+import { listenAbort, value } from '@orpc/shared'
 import type { HttpHandler, HttpRequestResolverExtras, PathParams, ResponseResolverInfo } from 'msw'
 import { http, passthrough } from 'msw'
 
@@ -163,12 +163,7 @@ export class HTTPProcedureUtils<
     return this.handler(
       ({ signal }) =>
         new Promise<never>((_, reject) => {
-          if (signal?.aborted) {
-            reject(signal.reason)
-            return
-          }
-
-          signal?.addEventListener('abort', () => reject(signal.reason), { once: true })
+          listenAbort(signal, reject)
         }),
     )
   }
diff --git a/packages/nest/src/implement.ts b/packages/nest/src/implement.ts
index f2df2cce6..ca156752d 100644
--- a/packages/nest/src/implement.ts
+++ b/packages/nest/src/implement.ts
@@ -36,6 +36,7 @@ import { StandardHandler } from '@orpc/server/standard'
 import type { Promisable } from '@orpc/shared'
 import {
   isAsyncIteratorObject,
+  listenAbort,
   mergeHttpPath,
   NullProtoObj,
   stringifyJSON,
@@ -431,11 +432,7 @@ function toStreamableFile(
   const type = flattenStandardHeader(headers['content-type'])
   headers['content-type'] = undefined
 
-  if (signal?.aborted) {
-    stream.destroy()
-  } else {
-    signal?.addEventListener('abort', () => stream.destroy(), { once: true })
-  }
+  listenAbort(signal, () => stream.destroy())
 
   return new StreamableFile(stream, { type })
 }
diff --git a/packages/node/src/batch-response-compression-handler-plugin.ts b/packages/node/src/batch-response-compression-handler-plugin.ts
index 29b50ca45..314503058 100644
--- a/packages/node/src/batch-response-compression-handler-plugin.ts
+++ b/packages/node/src/batch-response-compression-handler-plugin.ts
@@ -12,6 +12,7 @@ import type {
   StandardHandlerPlugin,
   StandardHandlerRoutingInterceptor,
 } from '@orpc/server/standard'
+import type { CompressionEncoding } from '@orpc/shared'
 import { toArray } from '@orpc/shared'
 import type { StandardBodyHint, StandardHeaders } from '@standard-server/core'
 import { flattenStandardHeader } from '@standard-server/core'
@@ -25,7 +26,7 @@ export interface BatchResponseCompressionHandlerPluginOptions {
    *
    * @default ['gzip', 'deflate']
    */
-  encodings?: readonly ('gzip' | 'deflate' | 'deflate-raw')[] | undefined
+  encodings?: readonly CompressionEncoding[] | undefined
 
   /**
    * The minimum response size in bytes required to trigger compression.
@@ -188,9 +189,7 @@ interface CompressionStreamPair {
  * as soon as the batch produces it instead of waiting in the compressor for the batch to end.
  * The few bytes each flush costs are what keep a streaming batch streaming.
  */
-function createFlushingCompressionStream(
-  encoding: Exclude<BatchResponseCompressionHandlerPluginOptions['encodings'], undefined>[number],
-): CompressionStreamPair {
+function createFlushingCompressionStream(encoding: CompressionEncoding): CompressionStreamPair {
   const zlibOptions = { flush: constants.Z_SYNC_FLUSH }
 
   const compressor =
diff --git a/packages/openapi/src/adapters/standard/openapi-link-codec.ts b/packages/openapi/src/adapters/standard/openapi-link-codec.ts
index 99851bc80..0c40a0461 100644
--- a/packages/openapi/src/adapters/standard/openapi-link-codec.ts
+++ b/packages/openapi/src/adapters/standard/openapi-link-codec.ts
@@ -93,7 +93,6 @@ export interface OpenAPILinkCodecOptions<T extends ClientContext> {
   errorResponseContentType?: string | undefined
 }
 
-const END_SLASH_REGEX = /\/$/
 const DOT_SEGMENT_REGEX = /(?:^|\/)\.{1,2}(?:\/|$)/
 
 export class OpenAPILinkCodec<T extends ClientContext> implements StandardLinkCodec<T> {
@@ -170,7 +169,7 @@ export class OpenAPILinkCodec<T extends ClientContext> implements StandardLinkCo
         data = Object.keys(remaining).length > 0 ? remaining : undefined
       }
 
-      pathname = `${basePathname.replace(END_SLASH_REGEX, '')}${pathname}` as `/${string}`
+      pathname = mergeHttpPath(basePathname, pathname)
 
       if (isBodylessMethod(method)) {
         const queryString = this.serializeQueryString(data, meta?.queryStyles)
@@ -236,7 +235,7 @@ export class OpenAPILinkCodec<T extends ClientContext> implements StandardLinkCo
       headers = mergeStandardHeaders(headers, serializeHeaders(input.headers, this.serializer))
     }
 
-    pathname = `${basePathname.replace(END_SLASH_REGEX, '')}${pathname}` as `/${string}`
+    pathname = mergeHttpPath(basePathname, pathname)
     const queryString = this.serializeQueryString(input?.query, meta?.queryStyles)
     const search = combineSearch(baseSearch, queryString)
     const url = `${pathname}${search ?? ''}${baseHash ?? ''}` as StandardUrl
diff --git a/packages/openapi/src/adapters/standard/openapi-matcher.ts b/packages/openapi/src/adapters/standard/openapi-matcher.ts
index aa1da7b08..76b24d001 100644
--- a/packages/openapi/src/adapters/standard/openapi-matcher.ts
+++ b/packages/openapi/src/adapters/standard/openapi-matcher.ts
@@ -8,7 +8,7 @@ import {
   walkProcedureContractsSync,
 } from '@orpc/server'
 import type { Value } from '@orpc/shared'
-import { mergeHttpPath, pathToHttpPath, value } from '@orpc/shared'
+import { mergeHttpPath, pathToHttpPath, stripHttpPathPrefix, value } from '@orpc/shared'
 
 import { DEFAULT_OPENAPI_METHOD } from '../../constants.ts'
 import { getOpenAPIMeta } from '../../meta.ts'
@@ -90,21 +90,13 @@ export class OpenAPIMatcher {
     | undefined
   > {
     if (prefix) {
-      if (!pathname.startsWith(prefix)) {
-        return undefined
-      }
+      const stripped = stripHttpPathPrefix(pathname, prefix)
 
-      const charAfterPrefix = pathname[prefix.length]
-
-      if (charAfterPrefix === '/') {
-        pathname = pathname.slice(prefix.length) as `/${string}`
-      } else if (charAfterPrefix === undefined) {
-        pathname = '/'
-      } else if (prefix[prefix.length - 1] === '/') {
-        pathname = pathname.slice(prefix.length - 1) as `/${string}`
-      } else {
+      if (stripped === undefined) {
         return undefined
       }
+
+      pathname = stripped
     }
 
     // most requests `await undefined` so conditionally await it to save a microtask turn
diff --git a/packages/openapi/src/helpers/form-data.ts b/packages/openapi/src/helpers/form-data.ts
index 12c5565c4..8312ec6b9 100644
--- a/packages/openapi/src/helpers/form-data.ts
+++ b/packages/openapi/src/helpers/form-data.ts
@@ -1,5 +1,5 @@
 import { isSchemaIssue } from '@orpc/contract'
-import { isNonPrimitive } from '@orpc/shared'
+import { isNonPrimitive, NON_NEGATIVE_INTEGER_REGEX } from '@orpc/shared'
 
 import { BracketNotationSerializer } from '../bracket-notation.ts'
 
@@ -97,7 +97,7 @@ export function getIssueMessage(error: unknown, path: string): string | undefine
       return issue.message
     }
 
-    if (path === '' && /^(?:0|[1-9]\d*)$/.test(issuePath)) {
+    if (path === '' && NON_NEGATIVE_INTEGER_REGEX.test(issuePath)) {
       return issue.message
     }
   }
diff --git a/packages/openapi/src/openapi-generator-components.ts b/packages/openapi/src/openapi-generator-components.ts
index f7dd8a9dc..b2b162f28 100644
--- a/packages/openapi/src/openapi-generator-components.ts
+++ b/packages/openapi/src/openapi-generator-components.ts
@@ -6,7 +6,7 @@ import {
   getJsonSchemaKeywordKind,
   mapJsonSchemaRefs,
 } from '@orpc/json-schema'
-import { getOwn, isDeepEqual, isPlainObject, setOwn } from '@orpc/shared'
+import { getOwn, isDeepEqual, isPlainObject, isPlainObjectOrArray, setOwn } from '@orpc/shared'
 
 import type { OpenAPIV3_2 } from './types.ts'
 
@@ -234,10 +234,6 @@ interface ReuseComparisonContext {
   visited: WeakMap<object, WeakSet<object>>
 }
 
-function isSchemaContainer(value: unknown): value is unknown[] | Record<PropertyKey, unknown> {
-  return Array.isArray(value) || isPlainObject(value)
-}
-
 /**
  * Where a value sits, mirroring how hoisting rewrites refs: only refs in subschemas are
  * rewritten, so literal data (`const`, `default`, `examples`, ...) is compared as is.
@@ -255,7 +251,7 @@ function areSchemasEquivalentForReuse(
   }
 
   // Only arrays and plain objects carry schema structure; primitives and values like Date are compared as a whole
-  if (!isSchemaContainer(candidate) || !isSchemaContainer(existing)) {
+  if (!isPlainObjectOrArray(candidate) || !isPlainObjectOrArray(existing)) {
     return isDeepEqual(candidate, existing)
   }
 
diff --git a/packages/pinia-colada/src/live-query.ts b/packages/pinia-colada/src/live-query.ts
index cdc53a9e7..6729d1d7b 100644
--- a/packages/pinia-colada/src/live-query.ts
+++ b/packages/pinia-colada/src/live-query.ts
@@ -1,5 +1,5 @@
 import type { Promisable } from '@orpc/shared'
-import { stringifyJSON, throwIfAborted } from '@orpc/shared'
+import { listenAbort, stringifyJSON, throwIfAborted } from '@orpc/shared'
 
 import type { SettleOnFirstDataOptions } from './stream-query.ts'
 import type { UseQueryFnContext } from './types.ts'
@@ -16,13 +16,9 @@ export function liveQuery<T>(
     const { entry, signal } = context
 
     // a fetch cut short (cancelled, superseded, or its last component left) reopens when used again
-    const markStale = () => {
+    listenAbort(signal, () => {
       entry.when = 0
-    }
-    if (signal.aborted) {
-      markStale()
-    }
-    signal.addEventListener('abort', markStale, { once: true })
+    })
 
     const stream = await queryFn(context)
 
diff --git a/packages/pinia-colada/src/stream-query.test.tsx b/packages/pinia-colada/src/stream-query.test.tsx
index feb49612d..eb2b53aa1 100644
--- a/packages/pinia-colada/src/stream-query.test.tsx
+++ b/packages/pinia-colada/src/stream-query.test.tsx
@@ -488,6 +488,30 @@ describe('serializableStreamedQuery', () => {
     await expect(queryCache.fetch(entry)).resolves.toMatchObject({ data: null })
   })
 
+  it('starts a new array when the entry data is cleared mid-stream', async () => {
+    const queryCache = createQueryCache()
+    const release = promiseWithResolvers<void>()
+    const entry = queryCache.ensure({
+      key: ['stream'],
+      query: serializableStreamedQuery(async () =>
+        (async function* () {
+          yield 1
+          entry.state.value = { status: 'pending', data: undefined, error: null } as any
+          yield 2
+          await release.promise
+        })(),
+      ),
+    })
+
+    const fetching = queryCache.fetch(entry)
+
+    // checked while the stream is still open, so the final write cannot be what makes it pass
+    await vi.waitFor(() => expect(entry.state.value.data).toEqual([2]), { interval: 1 })
+
+    release.resolve()
+    await expect(fetching).resolves.toMatchObject({ data: [2] })
+  })
+
   it('resolves with the streamed data when the entry data is cleared after the last chunk', async () => {
     const queryCache = createQueryCache()
     const entry = queryCache.ensure({
diff --git a/packages/pinia-colada/src/stream-query.ts b/packages/pinia-colada/src/stream-query.ts
index 824286f4e..e96c504a6 100644
--- a/packages/pinia-colada/src/stream-query.ts
+++ b/packages/pinia-colada/src/stream-query.ts
@@ -1,5 +1,5 @@
 import type { Promisable } from '@orpc/shared'
-import { throwIfAborted } from '@orpc/shared'
+import { listenAbort, takeLast, throwIfAborted } from '@orpc/shared'
 
 import type { UseQueryFnContext } from './types.ts'
 
@@ -61,13 +61,9 @@ export function serializableStreamedQuery<T>(
     }
 
     // a fetch cut short (cancelled, superseded, or its last component left) reopens when used again
-    const markStale = () => {
+    listenAbort(signal, () => {
       entry.when = 0
-    }
-    if (signal.aborted) {
-      markStale()
-    }
-    signal.addEventListener('abort', markStale, { once: true })
+    })
 
     const stream = await queryFn(context)
 
@@ -78,7 +74,7 @@ export function serializableStreamedQuery<T>(
     }
 
     const current = entry.state.value.data as T[] | undefined
-    let data = write(refetchMode === 'append' && current ? limitArraySize(current, maxChunks) : [])
+    let data = write(refetchMode === 'append' && current ? takeLast(current, maxChunks) : [])
 
     if (settleOnFirstData) {
       await stream[Symbol.asyncIterator]().return?.()
@@ -90,7 +86,7 @@ export function serializableStreamedQuery<T>(
 
       // read the entry, so writes made during the stream are kept
       const current = (entry.state.value.data as T[] | undefined) ?? []
-      data = write(limitArraySize([...current, chunk], maxChunks))
+      data = write(takeLast([...current, chunk], maxChunks))
     }
 
     // a stream that ends on abort must not be stored as fresh data
@@ -103,14 +99,6 @@ export function serializableStreamedQuery<T>(
     }
 
     // only an array is trimmed; another value written after the stream (e.g. `null`) is kept as is
-    return Array.isArray(cached) ? limitArraySize(cached, maxChunks) : cached
+    return Array.isArray(cached) ? takeLast(cached, maxChunks) : cached
   }
 }
-
-function limitArraySize<T>(items: T[], maxSize: number): T[] {
-  if (items.length <= maxSize) {
-    return items
-  }
-
-  return items.slice(items.length - maxSize)
-}
diff --git a/packages/server/src/adapters/standard/rpc-matcher.ts b/packages/server/src/adapters/standard/rpc-matcher.ts
index 2b6776cd0..89a02aa00 100644
--- a/packages/server/src/adapters/standard/rpc-matcher.ts
+++ b/packages/server/src/adapters/standard/rpc-matcher.ts
@@ -1,6 +1,6 @@
 import type { AnyProcedureContract } from '@orpc/contract'
 import type { Value } from '@orpc/shared'
-import { normalizeHttpPath, pathToHttpPath, value } from '@orpc/shared'
+import { normalizeHttpPath, pathToHttpPath, stripHttpPathPrefix, value } from '@orpc/shared'
 import type { StandardMethod } from '@standard-server/core'
 
 import { unlazy } from '../../lazy.ts'
@@ -126,21 +126,13 @@ export class RPCMatcher {
     }
 
     if (prefix) {
-      if (!pathname.startsWith(prefix)) {
-        return undefined
-      }
+      const stripped = stripHttpPathPrefix(pathname, prefix)
 
-      const charAfterPrefix = pathname[prefix.length]
-
-      if (charAfterPrefix === '/') {
-        pathname = pathname.slice(prefix.length) as `/${string}`
-      } else if (charAfterPrefix === undefined) {
-        pathname = '/'
-      } else if (prefix[prefix.length - 1] === '/') {
-        pathname = pathname.slice(prefix.length - 1) as `/${string}`
-      } else {
+      if (stripped === undefined) {
         return undefined
       }
+
+      pathname = stripped
     }
 
     if (pathname.length > 1 && pathname.endsWith('/')) {
diff --git a/packages/server/src/plugins/batch.ts b/packages/server/src/plugins/batch.ts
index 41f69a64d..7513ccf73 100644
--- a/packages/server/src/plugins/batch.ts
+++ b/packages/server/src/plugins/batch.ts
@@ -1,7 +1,13 @@
 import { ORPCError } from '@orpc/client'
 import { BATCH_HEADER, DEFAULT_BATCH_MAX_SIZE } from '@orpc/client/plugins'
 import type { Promisable, Value } from '@orpc/shared'
-import { isAsyncIteratorObject, promiseWithResolvers, toArray, value } from '@orpc/shared'
+import {
+  isAsyncIteratorObject,
+  listenAbort,
+  promiseWithResolvers,
+  toArray,
+  value,
+} from '@orpc/shared'
 import type { StandardHeaders, StandardLazyRequest, StandardResponse } from '@standard-server/core'
 import { flattenStandardHeader, parseStandardUrl } from '@standard-server/core'
 import type { ClientPeerSendMessage, ServerPeerSendMessage } from '@standard-server/peer'
@@ -521,19 +527,15 @@ export class BatchHandlerPlugin<T extends Context> implements StandardHandlerPlu
               .finally(() => resolve(undefined))
           }),
         )
-        const signal = interceptorOptions.request.signal
-        const closePeer = () => peer.close(signal?.reason)
-
-        if (signal?.aborted) {
-          void closePeer()
-        }
-
-        signal?.addEventListener('abort', closePeer)
+        const unlisten = listenAbort(
+          interceptorOptions.request.signal,
+          (reason) => void peer.close(reason),
+        )
 
         try {
           await promise
         } finally {
-          signal?.removeEventListener('abort', closePeer)
+          unlisten()
         }
       }
 
diff --git a/packages/server/src/plugins/request-compression.ts b/packages/server/src/plugins/request-compression.ts
index d95e97f20..21943ff21 100644
--- a/packages/server/src/plugins/request-compression.ts
+++ b/packages/server/src/plugins/request-compression.ts
@@ -1,5 +1,5 @@
 import { ORPCError } from '@orpc/client'
-import { toArray } from '@orpc/shared'
+import { isCompressionEncoding, parseContentEncodings, toArray } from '@orpc/shared'
 import { flattenStandardHeader } from '@standard-server/core'
 import { toFetchHeaders, toStandardBody } from '@standard-server/fetch'
 
@@ -42,7 +42,7 @@ export class RequestCompressionHandlerPlugin<
         flattenStandardHeader(interceptorOptions.request.headers['content-encoding']),
       )
 
-      if (encodings.length === 0 || !encodings.every(isSupportedEncoding)) {
+      if (encodings.length === 0 || !encodings.every(isCompressionEncoding)) {
         return next()
       }
 
@@ -94,22 +94,3 @@ export class RequestCompressionHandlerPlugin<
     }
   }
 }
-
-const SUPPORTED_ENCODINGS = ['gzip', 'deflate', 'deflate-raw'] as const
-type SupportedEncoding = (typeof SUPPORTED_ENCODINGS)[number]
-function isSupportedEncoding(encoding: string): encoding is SupportedEncoding {
-  return (SUPPORTED_ENCODINGS as readonly string[]).includes(encoding)
-}
-
-/**
- * Parse Content-Encoding into ordered codings (order applied).
- *
- * @see https://www.rfc-editor.org/rfc/rfc9110.html#name-content-encoding
- */
-function parseContentEncodings(header: string | undefined): string[] {
-  if (header === undefined) {
-    return []
-  }
-
-  return header.split(',').map((part) => part.trim().toLowerCase())
-}
diff --git a/packages/server/src/plugins/response-compression.ts b/packages/server/src/plugins/response-compression.ts
index c61953320..03a0b2f5d 100644
--- a/packages/server/src/plugins/response-compression.ts
+++ b/packages/server/src/plugins/response-compression.ts
@@ -1,4 +1,7 @@
+import type { CompressionEncoding } from '@orpc/shared'
 import {
+  estimateUtf8ByteLength,
+  estimateCompressibleFormDataSize,
   isAcceptableEncoding,
   isAsyncIteratorObject,
   isCompressibleContentType,
@@ -19,10 +22,6 @@ import type {
 } from '../adapters/standard/index.ts'
 import type { Context } from '../context.ts'
 
-// Rough UTF-8 estimate. Mostly ASCII text stays close to 1 byte/char;
-// occasional multi-byte characters increase the average.
-const AVG_BYTES_PER_CHAR = 1.2
-
 export interface ResponseCompressionHandlerPluginOptions<T extends Context> {
   /**
    * The compression schemes to use for response compression.
@@ -31,7 +30,7 @@ export interface ResponseCompressionHandlerPluginOptions<T extends Context> {
    *
    * @default ['gzip', 'deflate']
    */
-  encodings?: readonly ('gzip' | 'deflate' | 'deflate-raw')[] | undefined
+  encodings?: readonly CompressionEncoding[] | undefined
 
   /**
    * The minimum response size in bytes required to trigger compression.
@@ -165,32 +164,9 @@ export class ResponseCompressionHandlerPlugin<
           }
         }
       } else if (body instanceof FormData) {
-        const PART_OVERHEAD = 64 // approx bytes for boundary + Content-Disposition/Content-Type headers per part
-
-        let contentLength = 0
-        for (const [key, value] of body) {
-          contentLength += PART_OVERHEAD + key.length
-
-          if (value instanceof Blob) {
-            if (!Number.isFinite(value.size)) {
-              // Bun-s3 can use NaN for size
-              if (!this.isCompressibleContentType(value.type, interceptorOptions)) {
-                // Unknown non-compressible part size makes the estimate unreliable
-                contentLength = -Infinity
-                break
-              }
-
-              // Unknown size for compressible content - still apply compression
-              contentLength = Infinity
-            } else {
-              contentLength += this.isCompressibleContentType(value.type, interceptorOptions)
-                ? value.size
-                : -value.size
-            }
-          } else {
-            contentLength += value.length * AVG_BYTES_PER_CHAR
-          }
-        }
+        const contentLength = estimateCompressibleFormDataSize(body, (contentType) =>
+          this.isCompressibleContentType(contentType, interceptorOptions),
+        )
 
         if (contentLength >= this.threshold) {
           const res = new Response(body)
@@ -211,7 +187,7 @@ export class ResponseCompressionHandlerPlugin<
         }
       } else if (body instanceof URLSearchParams) {
         const string = body.toString()
-        if (string.length * AVG_BYTES_PER_CHAR >= this.threshold) {
+        if (estimateUtf8ByteLength(string) >= this.threshold) {
           const [, bodyHeaders] = toFetchBody(body, headers)
 
           return {
@@ -229,7 +205,7 @@ export class ResponseCompressionHandlerPlugin<
         }
       } else if (body !== undefined && !isAsyncIteratorObject(body)) {
         const [string, bodyHeaders] = toFetchBody(body, headers) as [string, StandardHeaders]
-        if (string.length * AVG_BYTES_PER_CHAR >= this.threshold) {
+        if (estimateUtf8ByteLength(string) >= this.threshold) {
           return {
             ...result,
             response: {
diff --git a/packages/server/src/plugins/signal-sync.test.ts b/packages/server/src/plugins/signal-sync.test.ts
index 9f06474bd..4a90661df 100644
--- a/packages/server/src/plugins/signal-sync.test.ts
+++ b/packages/server/src/plugins/signal-sync.test.ts
@@ -276,4 +276,68 @@ describe('signalSyncHandlerPlugin', () => {
     expect((await promise).status).toBe(500)
     expect(signal?.reason).toBe(reason)
   })
+
+  describe('interceptor', () => {
+    function intercept(output: unknown, signal: AbortSignal): Promise<any> {
+      const interceptor = new SignalSyncHandlerPlugin().init({}).clientInterceptors![0]!
+      return interceptor({ signal, next: async () => output } as any)
+    }
+
+    it('ignores a rejecting iterator return when the signal aborts', async () => {
+      const iterator = {
+        next: () => new Promise<never>(() => {}),
+        return: vi.fn(async () => {
+          throw new Error('return failed')
+        }),
+        [Symbol.asyncIterator]() {
+          return this
+        },
+      }
+
+      const controller = new AbortController()
+      await intercept(iterator, controller.signal)
+      controller.abort()
+
+      expect(iterator.return).toHaveBeenCalledTimes(1)
+      await new Promise((resolve) => setTimeout(resolve, 0)) // any unhandled rejection would fail the test
+    })
+
+    it('ignores a rejecting stream cancel when the signal aborts', async () => {
+      const cancel = vi.fn(() => {
+        throw new Error('cancel failed')
+      })
+
+      const controller = new AbortController()
+      const synced: ReadableStream = await intercept(
+        new ReadableStream({ cancel }),
+        controller.signal,
+      )
+      const reason = new Error('aborted')
+      controller.abort(reason)
+
+      await expect(synced.getReader().read()).rejects.toBe(reason)
+      expect(cancel).toHaveBeenCalledExactlyOnceWith(reason)
+      await new Promise((resolve) => setTimeout(resolve, 0)) // any unhandled rejection would fail the test
+    })
+
+    it('rejects a pending read with the abort reason when the signal aborts', async () => {
+      const pulled = promiseWithResolvers<void>()
+
+      const controller = new AbortController()
+      const synced: ReadableStream = await intercept(
+        new ReadableStream({ pull: () => pulled.resolve() }, { highWaterMark: 0 }),
+        controller.signal,
+      )
+
+      const reader = synced.getReader()
+      const read = reader.read()
+      await pulled.promise
+
+      const reason = new Error('aborted')
+      controller.abort(reason)
+
+      await expect(read).rejects.toBe(reason)
+      await new Promise((resolve) => setTimeout(resolve, 0)) // let the pending inner read settle
+    })
+  })
 })
diff --git a/packages/server/src/plugins/signal-sync.ts b/packages/server/src/plugins/signal-sync.ts
index 1214d4b6e..b77aca877 100644
--- a/packages/server/src/plugins/signal-sync.ts
+++ b/packages/server/src/plugins/signal-sync.ts
@@ -4,6 +4,7 @@ import {
   anyAbortSignal,
   AsyncIteratorClass,
   isAsyncIteratorObject,
+  listenAbort,
   override,
   runWithSignal,
   toArray,
@@ -52,16 +53,6 @@ export class SignalSyncHandlerPlugin<T extends Context> implements StandardHandl
   }
 }
 
-function listenAbort(signal: AbortSignal, onAbort: () => void): () => void {
-  if (signal.aborted) {
-    onAbort()
-  } else {
-    signal.addEventListener('abort', onAbort, { once: true })
-  }
-
-  return () => signal.removeEventListener('abort', onAbort)
-}
-
 function syncAsyncIterator(
   iterator: AsyncIterator<unknown, unknown>,
   signal: AbortSignal,
@@ -99,9 +90,9 @@ function syncReadableStream(
   return new ReadableStream(
     {
       start(controller) {
-        unlisten = listenAbort(signal, () => {
-          controller.error(signal.reason)
-          cancelling = reader.cancel(signal.reason)
+        unlisten = listenAbort(signal, (reason) => {
+          controller.error(reason)
+          cancelling = reader.cancel(reason)
           cancelling.catch(() => {}) // only awaited when the consumer cancels
         })
       },
diff --git a/packages/shared/src/array.test.ts b/packages/shared/src/array.test.ts
index 3cc467525..ebfd716b7 100644
--- a/packages/shared/src/array.test.ts
+++ b/packages/shared/src/array.test.ts
@@ -1,4 +1,4 @@
-import { splitInHalf, toArray } from './array.ts'
+import { splitInHalf, takeLast, toArray } from './array.ts'
 
 it('toArray', () => {
   expect(toArray(undefined)).toEqual([])
@@ -21,3 +21,12 @@ it('splitInHalf', () => {
   expect(splitInHalf([1])).toEqual([[1], []])
   expect(splitInHalf([])).toEqual([[], []])
 })
+
+it('takeLast', () => {
+  const items = [1, 2, 3]
+
+  expect(takeLast(items, 3)).toBe(items)
+  expect(takeLast(items, 5)).toBe(items)
+  expect(takeLast(items, 2)).toEqual([2, 3])
+  expect(takeLast(items, 0)).toEqual([])
+})
diff --git a/packages/shared/src/array.ts b/packages/shared/src/array.ts
index ba37808f0..5663f9a75 100644
--- a/packages/shared/src/array.ts
+++ b/packages/shared/src/array.ts
@@ -10,3 +10,14 @@ export function splitInHalf<T>(arr: readonly T[]): [T[], T[]] {
   const half = Math.ceil(arr.length / 2)
   return [arr.slice(0, half), arr.slice(half)]
 }
+
+/**
+ * Keeps only the last `maxSize` items, returning the same array when it already fits.
+ */
+export function takeLast<T>(items: T[], maxSize: number): T[] {
+  if (items.length <= maxSize) {
+    return items
+  }
+
+  return items.slice(items.length - maxSize)
+}
diff --git a/packages/shared/src/bigint.ts b/packages/shared/src/bigint.ts
index 9b90ec76d..b8548a6d9 100644
--- a/packages/shared/src/bigint.ts
+++ b/packages/shared/src/bigint.ts
@@ -1,13 +1,10 @@
+import { INTEGER_REGEX } from './regex.ts'
+
 /**
  * Default maximum digits, excluding a leading `-`, when parsing a `bigint` from a string.
  */
 export const DEFAULT_MAX_BIGINT_DIGITS = 4300
 
-/**
- * Matches `bigint.toString()` output only.
- */
-const BIGINT_PATTERN = /^-?(?:0|[1-9]\d*)$/
-
 /**
  * Parses a `bigint.toString()` string, or returns `undefined` if it is invalid or has more than `maxDigits` digits.
  * The length is checked first, since `BigInt()` is super-linear in the digit count.
@@ -15,7 +12,7 @@ const BIGINT_PATTERN = /^-?(?:0|[1-9]\d*)$/
 export function parseBigIntString(value: string, maxDigits: number): bigint | undefined {
   const digits = value.startsWith('-') ? value.length - 1 : value.length
 
-  if (digits > maxDigits || !BIGINT_PATTERN.test(value)) {
+  if (digits > maxDigits || !INTEGER_REGEX.test(value)) {
     return undefined
   }
 
diff --git a/packages/shared/src/http.test.ts b/packages/shared/src/http.test.ts
index 497d4be90..0fe3fd28a 100644
--- a/packages/shared/src/http.test.ts
+++ b/packages/shared/src/http.test.ts
@@ -1,13 +1,18 @@
 import {
+  estimateCompressibleFormDataSize,
+  estimateUtf8ByteLength,
   isAcceptableEncoding,
   isCompressibleContentType,
+  isCompressionEncoding,
   isNoTransformCacheControl,
   matchesHttpPath,
   matchesHttpPathPrefix,
   mergeHttpPath,
   normalizeHttpPath,
   parseAcceptEncodingQualities,
+  parseContentEncodings,
   pathToHttpPath,
+  stripHttpPathPrefix,
   varyByAcceptEncoding,
 } from './http.ts'
 
@@ -118,6 +123,20 @@ describe('matchesHttpPathPrefix', () => {
   })
 })
 
+describe('stripHttpPathPrefix', () => {
+  it.each([
+    ['/api/ping', '/api', '/ping'],
+    ['/api', '/api', '/'],
+    ['/api/', '/api', '/'],
+    ['/api/ping', '/api/', '/ping'],
+    ['/apis/ping', '/api', undefined],
+    ['/other', '/api', undefined],
+    ['/api?x', '/api', undefined],
+  ] as const)('%s without %s is %s', (pathname, prefix, expected) => {
+    expect(stripHttpPathPrefix(pathname, prefix)).toBe(expected)
+  })
+})
+
 describe('matchesHttpPath', () => {
   it.each([
     ['/api', '/api', true, 'matches exact path'],
@@ -240,6 +259,43 @@ describe('parseAcceptEncodingQualities', () => {
   })
 })
 
+describe('isCompressionEncoding', () => {
+  it.each([
+    ['gzip', true],
+    ['deflate', true],
+    ['deflate-raw', true],
+    ['br', false],
+    ['identity', false],
+    ['GZIP', false],
+  ])('%s: %s', (encoding, expected) => {
+    expect(isCompressionEncoding(encoding)).toBe(expected)
+  })
+})
+
+describe('parseContentEncodings', () => {
+  it('returns no codings when the header is missing', () => {
+    expect(parseContentEncodings(undefined)).toEqual([])
+  })
+
+  it('keeps the order codings were applied in, trimmed and lowercased', () => {
+    expect(parseContentEncodings(' GZIP , deflate')).toEqual(['gzip', 'deflate'])
+  })
+
+  it('ignores empty list elements', () => {
+    expect(parseContentEncodings('gzip,')).toEqual(['gzip'])
+    expect(parseContentEncodings(', gzip, , deflate')).toEqual(['gzip', 'deflate'])
+    expect(parseContentEncodings('')).toEqual([])
+  })
+})
+
+describe('estimateUtf8ByteLength', () => {
+  it('scales with the string length', () => {
+    expect(estimateUtf8ByteLength('')).toBe(0)
+    expect(estimateUtf8ByteLength('a'.repeat(10))).toBeGreaterThanOrEqual(10)
+    expect(estimateUtf8ByteLength('a'.repeat(20))).toBe(2 * estimateUtf8ByteLength('a'.repeat(10)))
+  })
+})
+
 describe('isAcceptableEncoding', () => {
   it.each([
     ['gzip', 'gzip', true],
@@ -289,3 +345,39 @@ describe('isNoTransformCacheControl', () => {
     expect(isNoTransformCacheControl('no-transform-extension')).toBe(false)
   })
 })
+
+describe('estimateCompressibleFormDataSize', () => {
+  const isCompressible = (type: string) => type.startsWith('text/')
+
+  it('counts strings, compressible blobs, and per-part overhead', () => {
+    const form = new FormData()
+    form.append('a', 'x'.repeat(10))
+    form.append('b', new Blob(['y'.repeat(100)], { type: 'text/plain' }))
+
+    expect(estimateCompressibleFormDataSize(form, isCompressible)).toBe(
+      64 + 1 + estimateUtf8ByteLength('x'.repeat(10)) + 64 + 1 + 100,
+    )
+  })
+
+  it('subtracts non-compressible blobs', () => {
+    const form = new FormData()
+    form.append('a', new Blob(['y'.repeat(100)], { type: 'image/png' }))
+
+    expect(estimateCompressibleFormDataSize(form, isCompressible)).toBe(64 + 1 - 100)
+  })
+
+  it('handles blobs with an unknown size', () => {
+    // a File appended without a filename is stored as is, so the patched size is what gets read
+    const compressible = new File(['a'], 'a.txt', { type: 'text/plain' })
+    Object.defineProperty(compressible, 'size', { value: Number.NaN })
+    const nonCompressible = new File(['a'], 'a.png', { type: 'image/png' })
+    Object.defineProperty(nonCompressible, 'size', { value: Number.NaN })
+
+    const form = new FormData()
+    form.append('a', compressible)
+    expect(estimateCompressibleFormDataSize(form, isCompressible)).toBe(Infinity)
+
+    form.append('b', nonCompressible)
+    expect(estimateCompressibleFormDataSize(form, isCompressible)).toBe(-Infinity)
+  })
+})
diff --git a/packages/shared/src/http.ts b/packages/shared/src/http.ts
index 48f21ab54..9be2c7ab4 100644
--- a/packages/shared/src/http.ts
+++ b/packages/shared/src/http.ts
@@ -35,6 +35,35 @@ export function matchesHttpPathPrefix(url: `/${string}`, prefix: `/${string}`):
   )
 }
 
+/**
+ * Removes `prefix` from `pathname`, or returns `undefined` when `pathname` is not under `prefix`.
+ * The rest of the path must start at a segment boundary, so `/apis` is not under `/api`.
+ */
+export function stripHttpPathPrefix(
+  pathname: `/${string}`,
+  prefix: `/${string}`,
+): `/${string}` | undefined {
+  if (!pathname.startsWith(prefix)) {
+    return undefined
+  }
+
+  const charAfterPrefix = pathname[prefix.length]
+
+  if (charAfterPrefix === '/') {
+    return pathname.slice(prefix.length) as `/${string}`
+  }
+
+  if (charAfterPrefix === undefined) {
+    return '/'
+  }
+
+  if (prefix[prefix.length - 1] === '/') {
+    return pathname.slice(prefix.length - 1) as `/${string}`
+  }
+
+  return undefined
+}
+
 export function matchesHttpPath(url: `/${string}`, path: `/${string}`): boolean {
   const pathWithoutEndSlash = path.endsWith('/') ? path.slice(0, path.length - 1) : path
 
@@ -52,6 +81,85 @@ export function matchesHttpPath(url: `/${string}`, path: `/${string}`): boolean
   return charAfterPrefix === undefined || charAfterPrefix === '?' || charAfterPrefix === '#'
 }
 
+/**
+ * Content codings that oRPC compression plugins can encode and decode.
+ */
+const COMPRESSION_ENCODINGS = ['gzip', 'deflate', 'deflate-raw'] as const
+
+export type CompressionEncoding = (typeof COMPRESSION_ENCODINGS)[number]
+
+export function isCompressionEncoding(encoding: string): encoding is CompressionEncoding {
+  return (COMPRESSION_ENCODINGS as readonly string[]).includes(encoding)
+}
+
+/**
+ * Parse Content-Encoding into ordered codings (order applied), ignoring empty list elements.
+ *
+ * @see https://www.rfc-editor.org/rfc/rfc9110.html#name-content-encoding
+ * @see https://www.rfc-editor.org/rfc/rfc9110.html#name-lists-rule-abnf-extension
+ */
+export function parseContentEncodings(header: string | undefined): string[] {
+  const codings: string[] = []
+
+  for (const part of header?.split(',') ?? []) {
+    const coding = part.trim().toLowerCase()
+
+    if (coding !== '') {
+      codings.push(coding)
+    }
+  }
+
+  return codings
+}
+
+// Rough UTF-8 estimate. Mostly ASCII text stays close to 1 byte/char;
+// occasional multi-byte characters increase the average.
+const AVG_UTF8_BYTES_PER_CHAR = 1.2
+
+/**
+ * Roughly estimates the UTF-8 byte length of a string without encoding it.
+ */
+export function estimateUtf8ByteLength(value: string): number {
+  return value.length * AVG_UTF8_BYTES_PER_CHAR
+}
+
+// approx bytes for boundary + Content-Disposition/Content-Type headers per part
+const FORM_DATA_PART_OVERHEAD = 64
+
+/**
+ * Estimates how many bytes of a FormData body are worth compressing. Non-compressible blobs count
+ * against the total. Returns `Infinity` when a compressible blob has an unknown size, and
+ * `-Infinity` when a non-compressible one does, since the estimate is then unreliable.
+ */
+export function estimateCompressibleFormDataSize(
+  formData: FormData,
+  isCompressible: (contentType: string) => boolean,
+): number {
+  let size = 0
+
+  for (const [key, value] of formData) {
+    size += FORM_DATA_PART_OVERHEAD + key.length
+
+    if (value instanceof Blob) {
+      if (!Number.isFinite(value.size)) {
+        // Bun-s3 can use NaN for size
+        if (!isCompressible(value.type)) {
+          return -Infinity
+        }
+
+        // Unknown size for compressible content - still apply compression
+        size = Infinity
+      } else {
+        size += isCompressible(value.type) ? value.size : -value.size
+      }
+    } else {
+      size += estimateUtf8ByteLength(value)
+    }
+  }
+
+  return size
+}
+
 const ACCEPT_ENCODING_QUALITY_REGEX = /^\s*q=([\d.]+)\s*$/i
 
 /**
diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts
index 91bed8782..98ba6ea02 100644
--- a/packages/shared/src/index.ts
+++ b/packages/shared/src/index.ts
@@ -15,6 +15,7 @@ export * from './plugin.ts'
 export * from './promise.ts'
 export * from './proxy.ts'
 export * from './queue.ts'
+export * from './regex.ts'
 export * from './signal.ts'
 export * from './stream.ts'
 export * from './tracing.ts'
diff --git a/packages/shared/src/object.test.ts b/packages/shared/src/object.test.ts
index c43e90946..7a07e7d42 100644
--- a/packages/shared/src/object.test.ts
+++ b/packages/shared/src/object.test.ts
@@ -12,6 +12,7 @@ import {
   getConstructors,
   getOwn,
   isPlainObject,
+  isPlainObjectOrArray,
   isPropertyKey,
   mergeTwoLevels,
   NullProtoObj,
@@ -807,3 +808,13 @@ describe('bindMethods', () => {
     expect(methods.double()).toBe(246)
   })
 })
+
+it('isPlainObjectOrArray', () => {
+  expect(isPlainObjectOrArray([])).toBe(true)
+  expect(isPlainObjectOrArray({})).toBe(true)
+  expect(isPlainObjectOrArray(Object.create(null))).toBe(true)
+  expect(isPlainObjectOrArray(new Date())).toBe(false)
+  expect(isPlainObjectOrArray(new Map())).toBe(false)
+  expect(isPlainObjectOrArray(null)).toBe(false)
+  expect(isPlainObjectOrArray('a')).toBe(false)
+})
diff --git a/packages/shared/src/object.ts b/packages/shared/src/object.ts
index 50ba1ee05..acaf1d92a 100644
--- a/packages/shared/src/object.ts
+++ b/packages/shared/src/object.ts
@@ -68,6 +68,15 @@ export function isPlainObject(value: unknown): value is Record<PropertyKey, unkn
   return proto === Object.prototype || !proto || !proto.constructor
 }
 
+/**
+ * Checks whether a value is an array or a plain object, i.e. something whose keys can be walked.
+ */
+export function isPlainObjectOrArray(
+  value: unknown,
+): value is unknown[] | Record<PropertyKey, unknown> {
+  return Array.isArray(value) || isPlainObject(value)
+}
+
 /**
  * Returns `object[key]` only when it is an own property, so keys like
  * `toString` do not resolve through the prototype chain.
diff --git a/packages/shared/src/regex.test.ts b/packages/shared/src/regex.test.ts
new file mode 100644
index 000000000..709f9a3ac
--- /dev/null
+++ b/packages/shared/src/regex.test.ts
@@ -0,0 +1,26 @@
+import { INTEGER_REGEX, NON_NEGATIVE_INTEGER_REGEX } from './regex.ts'
+
+it.each([
+  ['0', true],
+  ['42', true],
+  ['-42', true],
+  ['-0', true],
+  ['007', false],
+  ['1.5', false],
+  ['+1', false],
+  ['', false],
+  ['1e3', false],
+])('INTEGER_REGEX %s: %s', (value, expected) => {
+  expect(INTEGER_REGEX.test(value)).toBe(expected)
+})
+
+it.each([
+  ['0', true],
+  ['42', true],
+  ['-1', false],
+  ['01', false],
+  ['1.0', false],
+  ['', false],
+])('NON_NEGATIVE_INTEGER_REGEX %s: %s', (value, expected) => {
+  expect(NON_NEGATIVE_INTEGER_REGEX.test(value)).toBe(expected)
+})
diff --git a/packages/shared/src/regex.ts b/packages/shared/src/regex.ts
new file mode 100644
index 000000000..f261c4f4f
--- /dev/null
+++ b/packages/shared/src/regex.ts
@@ -0,0 +1,10 @@
+/**
+ * Matches an integer string: an optional `-`, then digits without leading zeros (`007` is rejected).
+ * Note that `-0` is accepted.
+ */
+export const INTEGER_REGEX = /^-?(?:0|[1-9]\d*)$/
+
+/**
+ * Matches a non-negative integer string without leading zeros, such as an array index.
+ */
+export const NON_NEGATIVE_INTEGER_REGEX = /^(?:0|[1-9]\d*)$/
diff --git a/packages/shared/src/signal.test.ts b/packages/shared/src/signal.test.ts
index ffa8a2259..453362420 100644
--- a/packages/shared/src/signal.test.ts
+++ b/packages/shared/src/signal.test.ts
@@ -1,6 +1,6 @@
 import { getEventListeners } from 'node:events'
 
-import { anyAbortSignal, disposableAllAbortSignal, runWithSignal } from './signal.ts'
+import { anyAbortSignal, disposableAllAbortSignal, listenAbort, runWithSignal } from './signal.ts'
 
 describe('disposableAllAbortSignal', () => {
   it('returns an undefined signal and a no-op dispose if contains undefined or empty array', () => {
@@ -250,3 +250,51 @@ describe('runWithSignal', () => {
     expect(removeSpy).toHaveBeenCalled()
   })
 })
+
+describe('listenAbort', () => {
+  it('calls the listener right away when already aborted', () => {
+    const controller = new AbortController()
+    controller.abort()
+    const onAbort = vi.fn()
+
+    listenAbort(controller.signal, onAbort)
+
+    expect(onAbort).toHaveBeenCalledTimes(1)
+    expect(onAbort).toHaveBeenCalledWith(controller.signal.reason)
+    expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0)
+  })
+
+  it('does nothing without a signal', () => {
+    const onAbort = vi.fn()
+
+    const remove = listenAbort(undefined, onAbort)
+    remove()
+
+    expect(onAbort).not.toHaveBeenCalled()
+  })
+
+  it('calls the listener once on abort', () => {
+    const controller = new AbortController()
+    const onAbort = vi.fn()
+
+    listenAbort(controller.signal, onAbort)
+    expect(onAbort).not.toHaveBeenCalled()
+
+    controller.abort('reason')
+    controller.abort()
+    expect(onAbort).toHaveBeenCalledTimes(1)
+    expect(onAbort).toHaveBeenCalledWith('reason')
+  })
+
+  it('returns a function that removes the listener', () => {
+    const controller = new AbortController()
+    const onAbort = vi.fn()
+
+    const remove = listenAbort(controller.signal, onAbort)
+    remove()
+    controller.abort()
+
+    expect(onAbort).not.toHaveBeenCalled()
+    expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0)
+  })
+})
diff --git a/packages/shared/src/signal.ts b/packages/shared/src/signal.ts
index d211a1a7f..a26e5c256 100644
--- a/packages/shared/src/signal.ts
+++ b/packages/shared/src/signal.ts
@@ -120,3 +120,28 @@ export async function runWithSignal<T>(
     }
   }
 }
+
+const noop = () => {}
+
+/**
+ * Calls `onAbort` with the abort reason once the signal aborts, or synchronously, before returning,
+ * if it is already aborted. Does nothing without a signal. Returns a function that removes the listener.
+ */
+export function listenAbort(
+  signal: AbortSignal | undefined,
+  onAbort: (reason: unknown) => void,
+): () => void {
+  if (!signal) {
+    return noop
+  }
+
+  const listener = () => onAbort(signal.reason)
+
+  if (signal.aborted) {
+    listener()
+  } else {
+    signal.addEventListener('abort', listener, { once: true })
+  }
+
+  return () => signal.removeEventListener('abort', listener)
+}
diff --git a/packages/swr/src/procedure-utils.test.ts b/packages/swr/src/procedure-utils.test.ts
index 8f604b37e..8517adf66 100644
--- a/packages/swr/src/procedure-utils.test.ts
+++ b/packages/swr/src/procedure-utils.test.ts
@@ -240,8 +240,36 @@ describe('procedureUtils', () => {
       expect(next).toHaveBeenCalledWith(new Error('__error__'))
     })
 
+    it('does not start the stream when unsubscribed before the call resolves', async () => {
+      const started = vi.fn()
+      const connected = promiseWithResolvers<void>()
+      client.mockImplementationOnce(async () => {
+        await connected.promise
+        return (async function* () {
+          started()
+          yield '__event__1'
+        })()
+      })
+      const subscriber = utils.subscriber({ refetchMode: 'reset' })
+
+      const next = vi.fn()
+      const unsubscribe = subscriber(key, { next })
+
+      unsubscribe()
+      connected.resolve()
+      await new Promise((resolve) => setTimeout(resolve, 10))
+
+      // no reset either, so a late resolution cannot clear the cache
+      expect(next).not.toHaveBeenCalled()
+      expect(started).not.toHaveBeenCalled()
+    })
+
     it('on error after unsubscribe', async () => {
+      const started = promiseWithResolvers<void>()
       client.mockImplementationOnce(async function* () {
+        yield 'started'
+        // resumed for the next value, so the first one has been handled
+        started.resolve()
         await new Promise((resolve) => setTimeout(resolve, 10))
         throw new Error('__error__')
       })
@@ -249,9 +277,12 @@ describe('procedureUtils', () => {
 
       const next = vi.fn()
       const unsubscribe = subscriber(key, { next })
+      // unsubscribe mid-stream, so the error reaches the subscriber after it was aborted
+      await started.promise
+      const callsBeforeUnsubscribe = next.mock.calls.length
       unsubscribe()
       await new Promise((resolve) => setTimeout(resolve, 20))
-      expect(next).toHaveBeenCalledTimes(0)
+      expect(next).toHaveBeenCalledTimes(callsBeforeUnsubscribe)
     })
 
     it('on non-AsyncIteratorObject output', async () => {
@@ -388,7 +419,11 @@ describe('procedureUtils', () => {
     })
 
     it('on error after unsubscribe', async () => {
+      const started = promiseWithResolvers<void>()
       client.mockImplementationOnce(async function* () {
+        yield 'started'
+        // resumed for the next value, so the first one has been handled
+        started.resolve()
         await new Promise((resolve) => setTimeout(resolve, 10))
         throw new Error('__error__')
       })
@@ -396,9 +431,12 @@ describe('procedureUtils', () => {
 
       const next = vi.fn()
       const unsubscribe = subscriber(key, { next })
+      // unsubscribe mid-stream, so the error reaches the subscriber after it was aborted
+      await started.promise
+      const callsBeforeUnsubscribe = next.mock.calls.length
       unsubscribe()
       await new Promise((resolve) => setTimeout(resolve, 20))
-      expect(next).toHaveBeenCalledTimes(0)
+      expect(next).toHaveBeenCalledTimes(callsBeforeUnsubscribe)
     })
 
     it('on non-AsyncIteratorObject output', async () => {
diff --git a/packages/swr/src/procedure-utils.ts b/packages/swr/src/procedure-utils.ts
index a39c5918a..5a1571991 100644
--- a/packages/swr/src/procedure-utils.ts
+++ b/packages/swr/src/procedure-utils.ts
@@ -1,6 +1,6 @@
 import type { Client, ClientContext } from '@orpc/client'
 import type { MaybeOptionalOptions } from '@orpc/shared'
-import { isAsyncIteratorObject, resolveMaybeOptionalOptions } from '@orpc/shared'
+import { isAsyncIteratorObject, resolveMaybeOptionalOptions, takeLast } from '@orpc/shared'
 
 import type { OperationKey, OperationKeyOptions, OperationKeyPrefixOptions } from './key.ts'
 import { generateOperationKey } from './key.ts'
@@ -131,11 +131,9 @@ export class ProcedureUtils<
             next(undefined, (old) => {
               const newData = Array.isArray(old) ? [...old, event] : [event]
 
-              if (typeof maxChunks === 'number' && newData.length > maxChunks) {
-                return newData.slice(newData.length - maxChunks) as InferSubscriberOutput<TOutput>
-              }
-
-              return newData as InferSubscriberOutput<TOutput>
+              return (
+                typeof maxChunks === 'number' ? takeLast(newData, maxChunks) : newData
+              ) as InferSubscriberOutput<TOutput>
             })
           }
         } catch (error) {
diff --git a/packages/tanstack-query/src/live-query.ts b/packages/tanstack-query/src/live-query.ts
index 3100ef686..f38883199 100644
--- a/packages/tanstack-query/src/live-query.ts
+++ b/packages/tanstack-query/src/live-query.ts
@@ -1,5 +1,5 @@
 import type { Promisable } from '@orpc/shared'
-import { stringifyJSON, throwIfAborted } from '@orpc/shared'
+import { listenAbort, stringifyJSON, throwIfAborted } from '@orpc/shared'
 import type { QueryFunction, QueryFunctionContext, QueryKey } from '@tanstack/query-core'
 
 import type { SettleOnFirstDataOptions } from './stream-query.ts'
@@ -13,11 +13,9 @@ export function liveQuery<TQueryFnData = unknown, TQueryKey extends QueryKey = Q
 
     // a stream cut short (cancelled, or its last observer left) reopens when observed again;
     // marked at abort, so the writes of a fetch that replaces this one clear it
-    const markStale = () => client.getQueryCache().find({ queryKey, exact: true })?.invalidate()
-    if (signal.aborted) {
-      markStale()
-    }
-    signal.addEventListener('abort', markStale, { once: true })
+    const unlisten = listenAbort(signal, () =>
+      client.getQueryCache().find({ queryKey, exact: true })?.invalidate(),
+    )
 
     try {
       const stream = await queryFn(context)
@@ -48,7 +46,7 @@ export function liveQuery<TQueryFnData = unknown, TQueryKey extends QueryKey = Q
 
       return last.chunk
     } finally {
-      signal.removeEventListener('abort', markStale)
+      unlisten()
     }
   }
 }
diff --git a/packages/tanstack-query/src/stream-query.test.ts b/packages/tanstack-query/src/stream-query.test.ts
index f93345160..cac41da28 100644
--- a/packages/tanstack-query/src/stream-query.test.ts
+++ b/packages/tanstack-query/src/stream-query.test.ts
@@ -362,6 +362,28 @@ describe('serializableStreamedQuery', () => {
       await promise1
     })
 
+    it('starts a new array when the cache is set to null mid-stream', async () => {
+      const key = ['null-mid-stream']
+      const release = promiseWithResolvers<void>()
+
+      const fetching = queryClient.fetchQuery({
+        queryKey: key,
+        queryFn: serializableStreamedQuery(async function* () {
+          yield 1
+          queryClient.setQueryData(key, null)
+          yield 2
+          await release.promise
+        }),
+      })
+
+      // checked while the stream is still open, so the final write cannot be what makes it pass
+      await vi.waitFor(() => expect(queryClient.getQueryData(key)).toEqual([2]), { interval: 1 })
+
+      release.resolve()
+      await expect(fetching).resolves.toEqual([2])
+      expect(queryClient.getQueryData(key)).toEqual([2])
+    })
+
     it('keeps a null written after the last chunk', async () => {
       const key = ['null-after-stream']
 
diff --git a/packages/tanstack-query/src/stream-query.ts b/packages/tanstack-query/src/stream-query.ts
index 4cd79e8be..e0a8057d3 100644
--- a/packages/tanstack-query/src/stream-query.ts
+++ b/packages/tanstack-query/src/stream-query.ts
@@ -1,5 +1,5 @@
 import type { Promisable } from '@orpc/shared'
-import { throwIfAborted } from '@orpc/shared'
+import { listenAbort, takeLast, throwIfAborted } from '@orpc/shared'
 import type { QueryFunction, QueryFunctionContext, QueryKey } from '@tanstack/query-core'
 
 export interface SerializableStreamedQueryOptions {
@@ -56,11 +56,9 @@ export function serializableStreamedQuery<
 
     // a stream cut short (cancelled, or its last observer left) reopens when observed again;
     // marked at abort, so the writes of a fetch that replaces this one clear it
-    const markStale = () => client.getQueryCache().find({ queryKey, exact: true })?.invalidate()
-    if (signal.aborted) {
-      markStale()
-    }
-    signal.addEventListener('abort', markStale, { once: true })
+    const unlisten = listenAbort(signal, () =>
+      client.getQueryCache().find({ queryKey, exact: true })?.invalidate(),
+    )
 
     try {
       const stream = await queryFn(context)
@@ -72,7 +70,7 @@ export function serializableStreamedQuery<
       }
 
       let data = client.setQueryData<TQueryFnData[]>(queryKey, (current) =>
-        refetchMode === 'append' && current ? limitArraySize(current, maxChunks) : [],
+        refetchMode === 'append' && current ? takeLast(current, maxChunks) : [],
       )!
 
       if (settleOnFirstData) {
@@ -83,9 +81,9 @@ export function serializableStreamedQuery<
       for await (const chunk of stream) {
         throwIfAborted(signal)
 
-        // read the cache, so writes made during the stream are kept
-        data = client.setQueryData<TQueryFnData[]>(queryKey, (current = []) =>
-          limitArraySize([...current, chunk], maxChunks),
+        // read the cache, so writes made during the stream are kept; a cleared entry (`null`) starts over
+        data = client.setQueryData<TQueryFnData[]>(queryKey, (current) =>
+          takeLast([...(current ?? []), chunk], maxChunks),
         )!
       }
 
@@ -96,17 +94,9 @@ export function serializableStreamedQuery<
       }
 
       // only an array is trimmed; another value written after the stream (e.g. `null`) is kept as is
-      return Array.isArray(cached) ? limitArraySize(cached, maxChunks) : cached
+      return Array.isArray(cached) ? takeLast(cached, maxChunks) : cached
     } finally {
-      signal.removeEventListener('abort', markStale)
+      unlisten()
     }
   }
 }
-
-function limitArraySize<T>(items: T[], maxSize: number): T[] {
-  if (items.length <= maxSize) {
-    return items
-  }
-
-  return items.slice(items.length - maxSize)
-}