fix(sdk): bound SSE frames and scan fragmented lines once

crates/tracedecay-sdk/BUILD.bazel

@@ -8,6 +8,8 @@ load("@rules_rust//cargo:defs.bzl", "cargo_build_script")
 load("@rules_rust//rust:defs.bzl", "rust_binary", "rust_library")
 load("//scripts:bazel/rust_test.bzl", "perf_rustc_flags", "rust_test")
 
+load("@aspect_rules_js//js:defs.bzl", "js_library")
+
 rust_library(
     name = "tracedecay-sdk",
     srcs = glob(["src/**/*.rs"]),
@@ -90,6 +92,12 @@ exports_files(["src/bin/generate.rs", "src/codegen.rs", "src/operations.rs"],
     visibility = ["//visibility:public"],
 )
 
+js_library(
+    name = "sse_framing_fixture",
+    srcs = ["tests/fixtures/sse-framing.json"],
+    visibility = ["//visibility:public"],
+)
+
 filegroup(
     name = "rustfmt_sources_2024",
     srcs = depset(glob(["src/**/*.rs"]) + glob(["tests/**/*.rs"], exclude = ["tests/fixtures/**"]) + ["//tests:support_sources"] + glob(["tests/sdk_suite/**/*.rs"]) + ["//tests:support_sources"]).to_list(),

crates/tracedecay-sdk/src/client.rs

@@ -26,8 +26,8 @@ const MAX_OPAQUE_BYTES: usize = 4_096;
 const MAX_REQUEST_ID_BYTES: usize = 512;
 /// Bound on one raw SSE line and on the event/id/data bytes retained for one
 /// frame before JSON decoding. Every canonical frame is a single JSON object of
-/// bounded metadata, request and operation identifiers (≤ 512 bytes), a
-/// sequence, a frontier with a 32-byte resume key, or a terminal receipt, so
+/// bounded metadata, request and operation identifiers (≤ 512 bytes), a test
+/// name (≤ 1 KiB), a sequence, a frontier or a terminal receipt, so
 /// 64 KiB leaves wide headroom while capping what a peer can make the client
 /// retain.
 const MAX_SSE_FRAME_BYTES: usize = 64 * 1024;

crates/tracedecay-sdk/tests/fixtures/sse-framing.json

@@ -0,0 +1,77 @@
+{
+  "limit_bytes": 65536,
+  "cases": [
+    {
+      "name": "data line including LF exceeds the limit",
+      "outcome": "oversized",
+      "prefix": "event: item\nid: 0\ndata: ",
+      "repeat": "a",
+      "count": 65530,
+      "suffix": "\n\n"
+    },
+    {
+      "name": "comment including LF exceeds the limit",
+      "outcome": "oversized",
+      "prefix": ":",
+      "repeat": "a",
+      "count": 65535,
+      "suffix": "\n\n"
+    },
+    {
+      "name": "unknown field including CRLF exceeds the limit",
+      "outcome": "oversized",
+      "prefix": "unknown: ",
+      "repeat": "a",
+      "count": 65526,
+      "suffix": "\r\n\r\n"
+    },
+    {
+      "name": "many small data fields exceed retained frame bytes",
+      "outcome": "oversized",
+      "prefix": "event: item\nid: 0\n",
+      "repeat": "data: abcdefghijklmnopqrstuvwxyz0123456789\n",
+      "count": 1800,
+      "suffix": "\n"
+    },
+    {
+      "name": "event value exceeds the line limit",
+      "outcome": "oversized",
+      "prefix": "event: ",
+      "repeat": "a",
+      "count": 65529,
+      "suffix": "\n\n"
+    },
+    {
+      "name": "id value exceeds the line limit",
+      "outcome": "oversized",
+      "prefix": "id: ",
+      "repeat": "a",
+      "count": 65532,
+      "suffix": "\n\n"
+    },
+    {
+      "name": "comment exactly at the raw LF line limit",
+      "outcome": "valid",
+      "prefix": ":",
+      "repeat": "a",
+      "count": 65534,
+      "suffix": "\n\n"
+    },
+    {
+      "name": "unknown field exactly at the raw CRLF line limit",
+      "outcome": "valid",
+      "prefix": "unknown: ",
+      "repeat": "a",
+      "count": 65525,
+      "suffix": "\r\n\r\n"
+    },
+    {
+      "name": "near-limit multiline UTF-8 data",
+      "outcome": "valid_open",
+      "prefix": "event: open\ndata: {\"event\":\"open\",\"data\":{\"correlation_id\":\"request.operation\",\ndata: \"padding\":\"",
+      "repeat": "🙂",
+      "count": 16320,
+      "suffix": "\"\ndata: ,\"frontier\":{\"next_sequence\":0,\"retained_from_sequence\":0,\"resume_token\":\"resume\"}}}\n\n"
+    }
+  ]
+}

crates/tracedecay-sdk/tests/sdk_suite/client.rs

@@ -715,6 +715,88 @@ const SSE_OPEN_FRAME: &str = concat!(
     "\"resume_token\":\"resume\"}}}\n\n"
 );
 
+#[test]
+fn sse_framing_policy_matches_the_typescript_client() {
+    #[derive(serde::Deserialize)]
+    struct FramingCase {
+        name: String,
+        outcome: String,
+        prefix: String,
+        repeat: String,
+        count: usize,
+        suffix: String,
+    }
+    #[derive(serde::Deserialize)]
+    struct FramingFixtures {
+        limit_bytes: usize,
+        cases: Vec<FramingCase>,
+    }
+    let fixtures: FramingFixtures =
+        serde_json::from_str(include_str!("../fixtures/sse-framing.json")).unwrap();
+    assert_eq!(fixtures.limit_bytes, SSE_FRAME_LIMIT_BYTES);
+    for fixture in fixtures.cases {
+        let fragment = format!(
+            "{}{}{}",
+            fixture.prefix,
+            fixture.repeat.repeat(fixture.count),
+            fixture.suffix
+        );
+        let body = if fixture.outcome == "valid_open" {
+            fragment
+        } else {
+            format!("{SSE_OPEN_FRAME}{fragment}")
+        };
+        if fixture.outcome == "oversized" {
+            let base_url = serve_abandonable_event_stream(body);
+            let client =
+                Client::builder(ConnectionMode::local(&base_url, "project.sdk", "sdk-token"))
+                    .build()
+                    .unwrap();
+            let mut stream = client
+                .stream_operation(
+                    "request.operation",
+                    StreamOptions {
+                        resume: None,
+                        max_reconnects: 3,
+                    },
+                )
+                .unwrap();
+            assert_eq!(stream.next().unwrap().unwrap().event, "open");
+            assert!(
+                matches!(
+                    stream.next(),
+                    Some(Err(ClientError::StreamFrameTooLarge {
+                        limit_bytes: SSE_FRAME_LIMIT_BYTES
+                    }))
+                ),
+                "{} must refuse before delivery",
+                fixture.name
+            );
+            assert!(
+                stream.next().is_none(),
+                "{} must not reconnect",
+                fixture.name
+            );
+        } else {
+            let body = format!("{body}{}", terminal_frame("completed", "completed"));
+            let (base_url, server) = serve(vec![event_response(&body)]);
+            let client =
+                Client::builder(ConnectionMode::local(&base_url, "project.sdk", "sdk-token"))
+                    .build()
+                    .unwrap();
+            let events = client
+                .stream_operation("request.operation", StreamOptions::default())
+                .unwrap()
+                .collect::<Result<Vec<_>, _>>()
+                .unwrap_or_else(|error| panic!("{}: {error}", fixture.name));
+            assert_eq!(events.len(), 2, "{}", fixture.name);
+            assert_eq!(events[0].event, "open", "{}", fixture.name);
+            assert!(events[1].terminal(), "{}", fixture.name);
+            server.join().unwrap();
+        }
+    }
+}
+
 #[test]
 fn oversized_sse_lines_and_frames_are_refused_without_delivery_or_reconnect() {
     let single_line = format!(

scripts/bazel/gen_builds.py

@@ -1025,6 +1025,8 @@ def unique_name(want):
             'load("@rules_rust//rust:defs.bzl", "rust_binary", "rust_library")\n'
             'load("//scripts:bazel/rust_test.bzl", "perf_rustc_flags", "rust_test")\n'
         )
+        if name == "tracedecay-sdk":
+            out.append('load("@aspect_rules_js//js:defs.bzl", "js_library")\n')
 
         if lib_t:
             for (vkind, closure), vfmap in sorted(
@@ -1387,6 +1389,12 @@ def unique_name(want):
                 'exports_files(["src/bin/generate.rs", "src/codegen.rs", "src/operations.rs"],',
                 '    visibility = ["//visibility:public"],',
                 ')\n',
+                # rules_js copies shared test data in its owning package.
+                'js_library(',
+                '    name = "sse_framing_fixture",',
+                '    srcs = ["tests/fixtures/sse-framing.json"],',
+                '    visibility = ["//visibility:public"],',
+                ')\n',
             ]
         out.extend(formatting_rules(dir_of[name], formatting_sources))
         return "\n".join(out)

sdks/typescript/BUILD.bazel

@@ -34,7 +34,7 @@ js_binary(
 
 vitest_bin.vitest_test(
     name = "unit_test",
-    data = [":pkg", ":node_modules"] + glob(["test/**", "*.json", "*.ts"]),
+    data = [":pkg", ":node_modules", "//crates/tracedecay-sdk:sse_framing_fixture"] + glob(["test/**", "*.json", "*.ts"]),
     args = ["run", "--exclude", "test/installed-package.test.ts"],
     chdir = package_name(),
 )

sdks/typescript/README.md

@@ -90,3 +90,20 @@ installing code. Prerelease package versions publish under the `beta`
 dist-tag; stable versions take `latest`. Setting `TRACEDECAY_SDK_TARBALL` to
 a prebuilt tarball path makes `test:installed` exercise that tarball instead
 of packing its own.
+
+## Event-stream framing limits
+
+The TypeScript and Rust clients use the same 64 KiB allowance for each raw SSE
+line, including its CR/LF ending, and for the UTF-8 event, ID and data bytes
+retained for a frame. Data-line separators count toward the frame allowance;
+replacement fields use the Rust client's conservative one-byte allowance.
+An inherited last-event ID stays charged until replaced or cleared. Canonical
+accepted-operation metadata, test-result names (at most 1 KiB), progress,
+frontiers, gaps and terminal receipts fit within this policy.
+
+The TypeScript client scans transport bytes once into a bounded line buffer,
+including when a transport supplies one large chunk. One byte of lookahead
+detects a line overflow. Oversized comments and unknown fields are refused too.
+Overflow throws `TraceDecayMalformedResponseError` with `payload.limitBytes`,
+cancels the response body and never reconnects or delivers the oversized frame.
+Transport-owned chunks are outside the parser's retained-state allowance.