Feat/sse outgoing by anjupathak03 · Pull Request #3961 · keploy/keploy

anjupathak03 · 2026-03-23T05:20:41Z

Describe the changes that are made

This pull request introduces support for handling HTTP streaming responses (such as Server-Sent Events and NDJSON) in the proxy layer. It implements streaming recording by offloading streamed response chunks to a file, and enables accurate replay of these streams during testing by reading and replaying the recorded chunks with original timing. The changes also generalize the handling of chunked responses and update the models to include references to external stream files.

HTTP Streaming Support and Recording:

Added logic to detect streaming responses (by inspecting Content-Type headers) and offload streamed response chunks to a .ndjson file during recording. Each chunk is timestamped and written as a JSON object, and a StreamRef is stored in the HTTP response model to reference the file. (pkg/agent/proxy/integrations/http/chunk.go, pkg/models/http.go, [1] [2] [3] [4] [5] [6] [7] [8] [9]
Updated the chunked response handling pipeline and related test cases to support the new streaming mechanism and pass the appropriate references and options. (pkg/agent/proxy/integrations/http/chunk.go, pkg/agent/proxy/integrations/http/chunk_test.go, [1] [2] [3] [4] [5] [6] [7]

Streaming Replay and Mocking:

Implemented replay logic for streaming responses in test mode: reads the recorded stream file, replays each chunk to the client with the original timing, and restores the Transfer-Encoding: chunked header for correct HTTP parsing. (pkg/agent/proxy/integrations/http/decode.go, pkg/agent/proxy/integrations/http/decode.goR161-L161)
Ensured that streamed responses are not redundantly parsed or read into memory, and that the StreamRef is propagated through the proxy pipeline. (pkg/agent/proxy/integrations/http/http.go, pkg/agent/proxy/integrations/http/http.goL149-R150)

General Improvements and Maintenance:

Improved error handling and logging for outgoing message recording, especially for network-closed errors. (pkg/agent/proxy/proxy.go, pkg/agent/proxy/proxy.goR826-R833)
Updated matcher logic to suppress success messages when failure logs are not emitted, supporting custom logging for streaming comparisons. (pkg/matcher/http/match.go, pkg/matcher/http/match.goL444-R446)
Minor fix to flag validation logic for disable-mapping in the CLI. (cli/provider/cmd.go, cli/provider/cmd.goR1058-R1066)

These changes provide robust support for capturing and replaying streaming HTTP responses, making the proxy more capable of handling modern web APIs that use streaming protocols.

Links & References

Closes: #[issue number that will be closed through this PR]

NA (if very small change like typo, linting, etc.)

🔗 Related PRs

NA

🐞 Related Issues

NA

📄 Related Documents

NA

What type of PR is this? (check all applicable)

Added e2e test pipeline?

👍 yes
🙅 no, because they aren't needed
🙋 no, because I need help

Added comments for hard-to-understand areas?

👍 yes
🙅 no, because the code is self-explanatory

Added to documentation?

📜 README.md
📓 Wiki
🙅 no documentation needed

Are there any sample code or steps to test the changes?

👍 yes, mentioned below
🙅 no, because it is not needed

Self Review done?

✅ yes
❌ no, because I need help

Any relevant screenshots, recordings or logs?

NA

🧠 Semantics for PR Title & Branch Name

Please ensure your PR title and branch name follow the Keploy semantics:

📌 PR Semantics Guide
📌 Branch Semantics Guide

Examples:

PR Title: fix: patch MongoDB document update bug
Branch Name: feat/#1-login-flow (You may skip mentioning the issue number in the branch name if the change is small and the PR description clearly explains it.)

Additional checklist:

Have you read the Contributing Guidelines on issues?
Have you followed the PR Semantics guide for naming this PR?
Have you followed the Branch Semantics guide for naming your branch?

Signed-off-by: Asish Kumar <[email protected]>

…noise configuration handling. Signed-off-by: Asish Kumar <[email protected]>

…ervice for SSE support. Signed-off-by: Asish Kumar <[email protected]>

…update replay service logic. Signed-off-by: Asish Kumar <[email protected]>

…ing the mock window and refine outgoing message recording logs. Signed-off-by: Asish Kumar <[email protected]>

Signed-off-by: Asish Kumar <[email protected]>

…le replay Signed-off-by: Asish Kumar <[email protected]>

Signed-off-by: Asish Kumar <[email protected]>

- Added a new function to merge consecutive SSE data fields, allowing for proper handling of multiline data. - Updated the `compareSSEFields` function to utilize the new merging logic. - Introduced tests to validate the normalization of multiline SSE data in both parsing and serialization contexts. Signed-off-by: Asish Kumar <[email protected]>

Signed-off-by: Asish Kumar <[email protected]>

Co-authored-by: kiloconnect[bot] <240665456+kiloconnect[bot]@users.noreply.github.com> Signed-off-by: Asish Kumar <[email protected]>

Signed-off-by: Asish Kumar <[email protected]>

…son logic for HTTP. Signed-off-by: Asish Kumar <[email protected]>

Signed-off-by: Asish Kumar <[email protected]>

Signed-off-by: Anju <[email protected]>

Signed-off-by: gouravkrosx <[email protected]>

- Introduced new types for HTTP streaming modes and configurations for better clarity and type safety. - Created a new function `SimulateHTTPStreaming` to handle streaming HTTP requests separately from standard HTTP requests. - Updated `SimulateHTTP` to utilize a shared request preparation function, reducing code duplication. - Enhanced streaming response comparison functions to return detailed mismatch information, improving error reporting. - Updated unit tests to use the new streaming simulation function and validate streaming responses more effectively.

Signed-off-by: gouravkrosx <[email protected]>

# Conflicts: # pkg/service/replay/replay.go # pkg/service/replay/utils_test.go

# Conflicts: # pkg/service/replay/replay.go

Signed-off-by: Anju <[email protected]>

Copilot

Pull request overview

Adds first-class support for HTTP streaming responses (SSE/NDJSON/etc.) across Keploy’s record → mock storage → replay pipeline, including new structured stream representations and replay-time stream comparison.

Changes:

Record: detect streaming/chunked responses and offload streamed chunks to streams/*.ndjson, storing a StreamRef in the mock.
Replay: rehydrate and replay streamed responses by reading the recorded stream file and emitting chunks with recorded timing; add streaming comparators for test-time validation.
Housekeeping: noise config deep-copy fixes, CLI flag validation tweak, and ownership restoration helpers for sudo-based runs.

Reviewed changes

Copilot reviewed 21 out of 21 changed files in this pull request and generated 12 comments.

Show a summary per file

File	Description
utils/permissions_windows.go	Adds Windows no-op for ownership restoration helper.
utils/permissions_unix.go	Adds `RestoreFileOwnership` helper for sudo scenarios.
pkg/util.go	Introduces streaming simulation + stream comparators and related helpers.
pkg/util_test.go	Adds unit tests for streaming simulation/comparison helpers.
pkg/service/replay/utils.go	Avoids mutating global noise config; adds streaming mock window helper and mapping upsert helper.
pkg/service/replay/utils_test.go	Adds tests for noise cloning/merge behavior and stream mock window.
pkg/service/replay/replay.go	Defers streaming tests to a sequential phase; wires streaming comparison into replay reporting.
pkg/service/replay/hooks.go	Routes HTTP streaming testcases through `SimulateHTTPStreaming`; adds noise map cloning helper.
pkg/service/record/record.go	Passes test-set config path through to agent options (for stream file resolution).
pkg/models/instrument.go	Extends `OutgoingOptions` with `ConfigPath` for resolving relative stream/assets.
pkg/models/http_stream_body.go	Adds structured stream chunk model and YAML marshal/unmarshal support for streaming bodies.
pkg/models/http_stream_body_test.go	Tests YAML marshal/unmarshal behavior for structured streaming bodies.
pkg/models/http.go	Adds `StreamRef` and internal `StreamBody` field to HTTP response model.
pkg/matcher/http/match.go	Suppresses success log when `emitFailureLogs` is disabled (streaming caller handles logging).
pkg/agent/proxy/proxy.go	Tweaks outgoing record error logging for network-closed cases; adds debug success log.
pkg/agent/proxy/integrations/http/http.go	Propagates `StreamRef` through HTTP mock parsing; avoids reading body when stream is offloaded.
pkg/agent/proxy/integrations/http/encode.go	Threads stream capture state through chunk handling and final mock creation.
pkg/agent/proxy/integrations/http/decode.go	Implements streaming mock replay by reading `StreamRef` ndjson and writing chunks to client.
pkg/agent/proxy/integrations/http/chunk.go	Adds stream file offload in chunked response handler and writes per-chunk ndjson records.
pkg/agent/proxy/integrations/http/chunk_test.go	Updates chunked response tests for new signature/streamRef plumbing.
cli/provider/cmd.go	Adjusts `disable-mapping` flag handling to avoid unintended overrides.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-03-23T05:26:05Z

pkg/agent/proxy/integrations/http/chunk.go

+		if err := os.MkdirAll(streamsDir, 0777); err != nil {
+			utils.LogError(h.Logger, err, "failed to create streams directory")


Creating the streams directory with mode 0777 is overly permissive (world-writable) and can be a security issue on multi-user systems. Consider using a tighter default (e.g., 0755) and relying on ownership restoration/umask as needed.

Suggested change

if err := os.MkdirAll(streamsDir, 0777); err != nil {

utils.LogError(h.Logger, err, "failed to create streams directory")

if err := os.MkdirAll(streamsDir, 0755); err != nil {

utils.LogError(h.Logger, err, "failed to create streams directory; verify permissions for configPath and parent directories")

Copilot · 2026-03-23T05:26:06Z

utils/permissions_unix.go

+// RestoreFileOwnership changes ownership of a file or directory to the original user
+// when running under sudo. This prevents the agent process (which runs as root via sudo)
+// from creating files that the CLI process (running as the normal user) cannot access.
+// Returns nil if not running under sudo or if ownership was successfully restored.


The doc comment for RestoreFileOwnership says "Returns nil" but the function doesn’t return anything. Please update the comment to match the signature/behavior (e.g., describe that it’s best-effort and logs debug on failures).

Suggested change

// RestoreFileOwnership changes ownership of a file or directory to the original user

// when running under sudo. This prevents the agent process (which runs as root via sudo)

// from creating files that the CLI process (running as the normal user) cannot access.

// Returns nil if not running under sudo or if ownership was successfully restored.

// RestoreFileOwnership changes ownership of files or directories to the original user

// when running under sudo. This prevents the agent process (which runs as root via sudo)

// from creating files that the CLI process (running as the normal user) cannot access.

// This is a best-effort helper: it does nothing when not running under sudo and logs debug

// messages if ownership restoration fails for any path.

Copilot · 2026-03-23T05:26:06Z

pkg/service/replay/replay.go

+					if !streamMatched {
+						r.logger.Error("streaming response mismatch detected", zap.String("testcase", tc.Name))
+						hadStreamingMismatch = true


This new error log doesn’t provide an actionable next step for the user (e.g., how to inspect the mismatch details or re-record the testcase). Please include a brief next action (like enabling debug logs / re-recording / inspecting streams) so failures are easier to remediate.

Copilot · 2026-03-23T05:26:07Z

pkg/util.go

+		aType := detectScalarType(strings.TrimSpace(a.value))
+		if eType != aType {
+			return false, "field-type mismatch"
+		}


compareSSEFields does not validate equality of non-"data" field values (e.g., id/event/retry). It only checks that the scalar types match, so "id:1" and "id:2" would compare equal. This undermines deterministic streaming comparisons; it should compare values (optionally with noise rules) rather than only types.

Suggested change

}

}

// For non-"data" fields, after confirming scalar type equality,

// ensure the actual values also match to maintain deterministic comparisons.

if strings.TrimSpace(e.value) != strings.TrimSpace(a.value) {

return false, "field-value mismatch"

}

Copilot · 2026-03-23T05:26:07Z

pkg/util.go

+			if len(eVal) != len(aVal) {
+				return false, "data-size mismatch"


For SSE "data" fields that are not JSON, the comparator only checks scalar type and string length (data-type mismatch / data-size mismatch) but not the actual value. This can allow mismatched payloads of the same length to pass. Consider comparing the actual normalized value (or a hash) rather than only size/type.

Suggested change

if len(eVal) != len(aVal) {

return false, "data-size mismatch"

if eVal != aVal {

return false, "data-value mismatch"

Copilot · 2026-03-23T05:26:08Z

pkg/agent/proxy/integrations/http/decode.go

+						utils.LogError(h.Logger, err, "failed to decode stream chunk")
+						break


In streaming replay, decoder.Decode errors (other than EOF) are logged but then the loop just breaks without reporting the error back via errCh. This can yield partial/corrupt responses while the caller still treats the mock as successful. Consider sending the error on errCh (and returning) so the test run can fail fast and surface the root cause.

Suggested change

utils.LogError(h.Logger, err, "failed to decode stream chunk")

break

utils.LogError(h.Logger, err, "failed to decode stream chunk; verify the recorded stream file is not truncated or corrupted and matches the expected format", zap.String("path", filePath))

errCh <- err

return

Copilot · 2026-03-23T05:26:08Z

pkg/agent/proxy/integrations/http/decode.go

+						time.Sleep(diff)
+					}
+					prevTS = chunk.TS
+
+					for _, field := range chunk.Data {
+						if field.Key == "raw" {
+							if _, err := clientConn.Write([]byte(field.Value)); err != nil {
+								utils.LogError(h.Logger, err, "failed to write chunk to client")


Streaming replay uses time.Sleep(diff) to preserve inter-chunk timing, but it doesn’t respect ctx cancellation/timeouts. If the recorded diff is large (or the test is canceled), this goroutine will keep sleeping and can delay shutdown. Consider replacing sleep with a timer + select on ctx.Done().

Suggested change

time.Sleep(diff)

}

prevTS = chunk.TS

for _, field := range chunk.Data {

if field.Key == "raw" {

if _, err := clientConn.Write([]byte(field.Value)); err != nil {

utils.LogError(h.Logger, err, "failed to write chunk to client")

timer := time.NewTimer(diff)

select {

case <-ctx.Done():

// Ensure the timer is cleaned up to avoid leaks.

if !timer.Stop() {

select {

case <-timer.C:

default:

}

}

// Log with a clear next step for the user.

utils.LogError(h.Logger, ctx.Err(), "streaming replay canceled while waiting between chunks; verify test timeouts or context cancellation configuration")

errCh <- ctx.Err()

return

case <-timer.C:

// proceed with sending the next chunk

}

}

prevTS = chunk.TS

for _, field := range chunk.Data {

if field.Key == "raw" {

if _, err := clientConn.Write([]byte(field.Value)); err != nil {

utils.LogError(h.Logger, err, "failed to write chunk to client; ensure the client is still connected and retry the test if needed")

Copilot · 2026-03-23T05:26:08Z

pkg/util.go

+
+	if len(expected.body) != len(actual.body) {
+		return false, "body-size mismatch"
+	}


compareMultipartPart falls back to comparing only body size for non-JSON parts. That means different text parts with the same length will be considered equal, which can hide real mismatches in multipart streams. Consider comparing the normalized body bytes/strings for text/* (and possibly XML) parts, not just length.

Suggested change

}

}

if !bytes.Equal(expected.body, actual.body) {

return false, "body mismatch"

}

Copilot · 2026-03-23T05:26:08Z

pkg/agent/proxy/integrations/http/chunk.go

+	if isStream {
+		configPath := opts.ConfigPath
+		if configPath == "" {
+			configPath = "."
+		}
+
+		streamsDir := filepath.Join(configPath, "streams")
+		if err := os.MkdirAll(streamsDir, 0777); err != nil {
+			utils.LogError(h.Logger, err, "failed to create streams directory")
+			return err
+		}
+		// Restore ownership of dirs created by the agent (root) so the CLI process
+		// (running as the normal user) can write mocks/tests into the same test-set dir.
+		utils.RestoreFileOwnership(h.Logger, configPath, streamsDir)
+
+		streamPath = filepath.Join("streams", fmt.Sprintf("stream_%d_%d.ndjson", time.Now().UnixNano(), pUtil.GetNextID()))
+		fullPath := filepath.Join(configPath, streamPath)
+
+		f, err := os.Create(fullPath)
+		if err != nil {
+			utils.LogError(h.Logger, err, "failed to create stream file")
+			return err
+		}
+		defer f.Close()
+		file = f
+		encoder = json.NewEncoder(file)
+		utils.RestoreFileOwnership(h.Logger, fullPath)
+		h.Logger.Debug("Created stream file", zap.String("path", fullPath))
+	}


When isStream=true, chunkedResponse writes only the subsequent reads from destConn into the stream file. Any chunk/body bytes that were already read into resp/finalResp before calling handleChunkedResponses (pUtil.ReadBytes can include body data) are never persisted, so replay can miss the beginning of the stream. Consider extracting any bytes after the header terminator from finalResp and writing them to the stream file before entering the read loop (or passing the initial body bytes explicitly).

Copilot · 2026-03-23T05:26:09Z

pkg/util.go

+		if len(line) != len(expected) {
+			logger.Debug("plain-text stream mismatch",
+				zap.Int("frame_index", nextExpected),
+				zap.Int("expected_size", len(expected)),
+				zap.Int("actual_size", len(line)))
+			mismatchInfo := &StreamMismatchInfo{
+				FrameIndex:    nextExpected,
+				ExpectedFrame: expected,
+				ActualFrame:   line,
+				Reason:        fmt.Sprintf("size mismatch: expected %d bytes, got %d bytes", len(expected), len(line)),
+			}


comparePlainTextStream currently treats frames as matching as long as their byte lengths are equal; it never compares the actual content. This can produce false positives (different text with same length will pass). Consider comparing normalized line contents (or JSON-compare where applicable) and only using size checks as an optimization/fallback.

officialasishkumar and others added 30 commits February 20, 2026 01:51

feat: add http and sse streaming support

41b72c1

Signed-off-by: Asish Kumar <[email protected]>

feat: Add support for asynchronous streaming HTTP replay and improve …

c057e23

…noise configuration handling. Signed-off-by: Asish Kumar <[email protected]>

feat: Add SSE Redis application with Keploy tests and update replay s…

5c19a6d

…ervice for SSE support. Signed-off-by: Asish Kumar <[email protected]>

feat: Add new SSE Redis application with Keploy tests and mocks, and …

2a6ca1a

…update replay service logic. Signed-off-by: Asish Kumar <[email protected]>

feat: Optimize mock filter updates for async streaming replay by pinn…

becfff8

…ing the mock window and refine outgoing message recording logs. Signed-off-by: Asish Kumar <[email protected]>

Merge branch 'main' into feat/sse-support

a01b6a5

fix: handle copilot comments

95707d7

Signed-off-by: Asish Kumar <[email protected]>

feat: use api timeout

7052723

Signed-off-by: Asish Kumar <[email protected]>

Merge branch 'main' into feat/sse-support

c7aa7d8

feat: change the response body structure

a7211aa

Signed-off-by: Asish Kumar <[email protected]>

Revert last commit

a5e970d

feat: migration to structured streaming bodies with backward-compatib…

19f8440

…le replay Signed-off-by: Asish Kumar <[email protected]>

feat: migration to structured streaming bodies for text/plain

074074a

Signed-off-by: Asish Kumar <[email protected]>

feat: add sse redis app

7df72f4

Signed-off-by: Asish Kumar <[email protected]>

Merge branch 'main' into feat/sse-support

048c88d

feat: remove pipeline

4989d6f

Signed-off-by: Asish Kumar <[email protected]>

fix: preserve recorded inter-request timing

4a4becf

Signed-off-by: Asish Kumar <[email protected]>

fix: change warn to debug

b454a15

Signed-off-by: Asish Kumar <[email protected]>

fix: change info to debug

24f6354

Signed-off-by: Asish Kumar <[email protected]>

fix: nil check

54d3834

Signed-off-by: Asish Kumar <[email protected]>

fix: nil check

50a92e7

Signed-off-by: Asish Kumar <[email protected]>

fix: add next steps

de25d6e

Signed-off-by: Asish Kumar <[email protected]>

refactor: update JSON unmarshaling to use json.NewDecoder

54f5be9

Signed-off-by: Asish Kumar <[email protected]>

fix: add next step in erro

d0d07db

Co-authored-by: kiloconnect[bot] <240665456+kiloconnect[bot]@users.noreply.github.com> Signed-off-by: Asish Kumar <[email protected]>

fix: add next step in error

b66745e

Co-authored-by: kiloconnect[bot] <240665456+kiloconnect[bot]@users.noreply.github.com> Signed-off-by: Asish Kumar <[email protected]>

fix: remove extra sorting

ca56f17

Signed-off-by: Asish Kumar <[email protected]>

feat: implement async HTTP request handling and improve simulation hooks

49d4959

Signed-off-by: Asish Kumar <[email protected]>

refactor: simplify inter-request timing logic in RunTestSet function

e27a8d9

Signed-off-by: Asish Kumar <[email protected]>

feat: Add inter-request timing enforcement and streaming body compari…

d818d72

…son logic for HTTP. Signed-off-by: Asish Kumar <[email protected]>

officialasishkumar and others added 25 commits February 26, 2026 10:28

fix: remove unit tests

5b0b71d

Signed-off-by: Asish Kumar <[email protected]>

fix: add comments

02e60f3

Signed-off-by: Asish Kumar <[email protected]>

fix: add comments

09a01c3

Signed-off-by: Asish Kumar <[email protected]>

refactor

08a2c84

Signed-off-by: Asish Kumar <[email protected]>

refactor

a3d5dd0

Signed-off-by: Asish Kumar <[email protected]>

refactor

386524d

Signed-off-by: Asish Kumar <[email protected]>

refactor

7a16529

Signed-off-by: Asish Kumar <[email protected]>

feat: defer streaming tests to run sequentially after normal tests run

99ebb76

Signed-off-by: Anju <[email protected]>

Merge branch 'main' into feat/sse-support

a414525

fix: lost changes for saving streaming body

234aa48

fix: text/plain chunking identification

5ade6ac

Signed-off-by: gouravkrosx <[email protected]>

chore: logging

47da178

Signed-off-by: gouravkrosx <[email protected]>

Merge branch 'main' into feat/sse-support

384cb5f

remove unused code

069f6c7

Signed-off-by: gouravkrosx <[email protected]>

remove unused code

5b7f7ef

Signed-off-by: gouravkrosx <[email protected]>

fix: test-mock mappings

5e57293

Signed-off-by: gouravkrosx <[email protected]>

fix: build

e518595

Signed-off-by: gouravkrosx <[email protected]>

fix: lint and go test

c94a828

Signed-off-by: gouravkrosx <[email protected]>

Merge remote-tracking branch 'origin/main' into feat/sse-support

d28167a

# Conflicts: # pkg/service/replay/replay.go # pkg/service/replay/utils_test.go

Merge branch 'main' into feat/sse-support

272b73d

Merge branch 'main' into feat/sse-support

9724fa1

Merge remote-tracking branch 'origin/main' into feat/sse-support

3129504

# Conflicts: # pkg/service/replay/replay.go

Merge branch 'main' into feat/sse-support

62cc1f6

feat: add sse outgoin

f2e9f10

Signed-off-by: Anju <[email protected]>

anjupathak03 requested review from Sarthak160 and gouravkrosx as code owners March 23, 2026 05:20

Copilot AI review requested due to automatic review settings March 23, 2026 05:20

Copilot started reviewing on behalf of anjupathak03 March 23, 2026 05:21 View session

Copilot AI reviewed Mar 23, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Feat/sse outgoing#3961

Feat/sse outgoing#3961
anjupathak03 wants to merge 67 commits intomainfrom
feat/sse-outgoing

anjupathak03 commented Mar 23, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Mar 23, 2026

Uh oh!

Copilot AI Mar 23, 2026

Uh oh!

Copilot AI Mar 23, 2026

Uh oh!

Copilot AI Mar 23, 2026

Uh oh!

Copilot AI Mar 23, 2026

Uh oh!

Copilot AI Mar 23, 2026

Uh oh!

Copilot AI Mar 23, 2026

Uh oh!

Copilot AI Mar 23, 2026

Uh oh!

Copilot AI Mar 23, 2026

Uh oh!

Copilot AI Mar 23, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

		if err := os.MkdirAll(streamsDir, 0777); err != nil {
		utils.LogError(h.Logger, err, "failed to create streams directory")

-		}
+		}
+		// For non-"data" fields, after confirming scalar type equality,
+		// ensure the actual values also match to maintain deterministic comparisons.
+		if strings.TrimSpace(e.value) != strings.TrimSpace(a.value) {
+			return false, "field-value mismatch"
+		}

		if len(eVal) != len(aVal) {
		return false, "data-size mismatch"

		utils.LogError(h.Logger, err, "failed to decode stream chunk")
		break

-						utils.LogError(h.Logger, err, "failed to decode stream chunk")
-						break
+						utils.LogError(h.Logger, err, "failed to decode stream chunk; verify the recorded stream file is not truncated or corrupted and matches the expected format", zap.String("path", filePath))
+						errCh <- err
+						return

-						time.Sleep(diff)
-					}
-					prevTS = chunk.TS
-					for _, field := range chunk.Data {
-						if field.Key == "raw" {
-							if _, err := clientConn.Write([]byte(field.Value)); err != nil {
-								utils.LogError(h.Logger, err, "failed to write chunk to client")
+						timer := time.NewTimer(diff)
+						select {
+						case <-ctx.Done():
+							// Ensure the timer is cleaned up to avoid leaks.
+							if !timer.Stop() {
+								select {
+								case <-timer.C:
+								default:
+								}
+							}
+							// Log with a clear next step for the user.
+							utils.LogError(h.Logger, ctx.Err(), "streaming replay canceled while waiting between chunks; verify test timeouts or context cancellation configuration")
+							errCh <- ctx.Err()
+							return
+						case <-timer.C:
+							// proceed with sending the next chunk
+						}
+					}
+					prevTS = chunk.TS
+					for _, field := range chunk.Data {
+						if field.Key == "raw" {
+							if _, err := clientConn.Write([]byte(field.Value)); err != nil {
+								utils.LogError(h.Logger, err, "failed to write chunk to client; ensure the client is still connected and retry the test if needed")

-	}
+	}
+	if !bytes.Equal(expected.body, actual.body) {
+		return false, "body mismatch"
+	}

Uh oh!

Conversation

anjupathak03 commented Mar 23, 2026

Describe the changes that are made

Links & References

🔗 Related PRs

🐞 Related Issues

📄 Related Documents

What type of PR is this? (check all applicable)

Added e2e test pipeline?

Added comments for hard-to-understand areas?

Added to documentation?

Are there any sample code or steps to test the changes?

Self Review done?

Any relevant screenshots, recordings or logs?

🧠 Semantics for PR Title & Branch Name

Additional checklist:

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants