From 1001dcb56957a2b62ccaf3e4e2a812a38b1b8564 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Sun, 14 Jun 2026 21:50:03 -0400 Subject: [PATCH] fix(e2e): demultiplex act container stream for reliable CI output parsing Raw io.Copy left Docker multiplexed-stream headers in captured act logs on CI Linux (no TTY allocated), corrupting conclusion, job-failure, infra-saturation, and crash parsing. Demultiplex via stdcopy.StdCopy with a raw-stream fallback. Adds a synthetic multiplexed-stream unit test. Signed-off-by: Joshua Temple --- e2e/harness/act.go | 132 ++++++++++++++++++++++++---------------- e2e/harness/act_test.go | 106 ++++++++++++++++++++++++++++++++ 2 files changed, 186 insertions(+), 52 deletions(-) diff --git a/e2e/harness/act.go b/e2e/harness/act.go index 1d00d808..c76e3db8 100644 --- a/e2e/harness/act.go +++ b/e2e/harness/act.go @@ -3,11 +3,13 @@ package harness import ( "bytes" "context" + "encoding/binary" "encoding/json" "fmt" "io" "strings" + "github.com/moby/moby/api/pkg/stdcopy" "github.com/moby/moby/api/types/container" "github.com/moby/moby/api/types/mount" "github.com/testcontainers/testcontainers-go" @@ -269,23 +271,22 @@ func (a *ActRunner) RunWorkflow(ctx context.Context, opts RunOpts) (*ExtendedWor return nil, fmt.Errorf("failed to run act: %w", err) } - // Read logs from the reader - var logs bytes.Buffer - if reader != nil { - _, err = io.Copy(&logs, reader) - if err != nil { - return nil, fmt.Errorf("failed to read act output: %w", err) - } + // Read and demultiplex the container's hijacked-attach stream. Without + // demuxing, a TTY-less exec (CI Linux) prefixes each chunk with Docker's + // 8-byte stream header, which corrupts the JSON the parser consumes. + logs, err := readDemuxedStream(reader) + if err != nil { + return nil, fmt.Errorf("failed to read act output: %w", err) } // Parse JSON output - result, err := ParseActOutput(logs.String()) + result, err := ParseActOutput(logs) if err != nil { // Fall back to basic result if parsing fails result = &ExtendedWorkflowResult{ Conclusion: "success", Jobs: make(map[string]*JobResultExtended), - Logs: logs.String(), + Logs: logs, } } @@ -352,21 +353,21 @@ func (a *ActRunner) RunWorkflowFromRepo(ctx context.Context, opts RunOpts) (*Ext return nil, fmt.Errorf("failed to run act: %w", err) } - var logs bytes.Buffer - if reader != nil { - _, _ = io.Copy(&logs, reader) + // Read and demultiplex the container's hijacked-attach stream so the JSON + // the parser consumes is free of Docker's per-chunk stream headers. On a + // TTY-less exec (CI Linux) those headers would otherwise corrupt the + // conclusion, job-failure, infra-saturation, and crash detection. + cleanedLogs, err := readDemuxedStream(reader) + if err != nil { + return nil, fmt.Errorf("failed to read act output: %w", err) } - // Clean the output: strip Docker's multiplexed stream headers - // Docker exec output contains 8-byte headers for each chunk - cleanedLogs := stripDockerStreamHeaders(logs.String()) - result, err := ParseActOutput(cleanedLogs) if err != nil { result = &ExtendedWorkflowResult{ Conclusion: "success", Jobs: make(map[string]*JobResultExtended), - Logs: logs.String(), + Logs: cleanedLogs, } // The parse fallback bypasses ParseActOutput's crash detection, so run // it directly here against the raw logs: a crash dump is the most likely @@ -491,41 +492,68 @@ func (a *ActRunner) buildActArgs(opts RunOpts, eventPath string) string { return args } -// stripDockerStreamHeaders removes Docker's multiplexed stream headers from output. -// Docker exec output contains 8-byte headers for each chunk: -// - 1 byte: stream type (0=stdin, 1=stdout, 2=stderr) -// - 3 bytes: padding -// - 4 bytes: payload size (big endian) -func stripDockerStreamHeaders(input string) string { - var result bytes.Buffer - data := []byte(input) - - for i := 0; i < len(data); { - // Look for the start of a JSON object - if data[i] == '{' { - // Find the end of this JSON line (newline or end of data) - end := i - braceCount := 0 - for end < len(data) { - if data[end] == '{' { - braceCount++ - } else if data[end] == '}' { - braceCount-- - if braceCount == 0 { - end++ - break - } - } - end++ - } - result.Write(data[i:end]) - result.WriteByte('\n') - i = end - } else { - // Skip non-JSON bytes (headers, control chars, etc.) - i++ - } +// dockerStreamHeaderLen is the size, in bytes, of the header Docker prepends to +// each frame of a multiplexed (non-TTY) hijacked-attach stream: +// - 1 byte: stream type (0=stdin, 1=stdout, 2=stderr) +// - 3 bytes: padding (always zero) +// - 4 bytes: payload size (big-endian uint32) +const dockerStreamHeaderLen = 8 + +// readDemuxedStream drains a container exec's hijacked-attach reader and returns +// clean text with Docker's per-frame stream headers removed. +// +// A container created without a TTY (every act run here, and notably CI Linux +// where Docker allocates no TTY) returns a MULTIPLEXED stream: each chunk is +// prefixed with an 8-byte header (stream type + big-endian length). A raw +// io.Copy leaves those headers interspersed in the captured logs, which lands a +// `\x02\x00...` prefix on JSON lines and corrupts conclusion, job-failure, +// infra-saturation, and crash parsing. Docker Desktop happens to demux for us, +// which is why the corruption never reproduces locally and only bites in CI. +// +// We demultiplex with stdcopy.StdCopy (the same primitive testcontainers' +// exec.Multiplexed uses), merging stdout and stderr into a single buffer; act +// writes its --json log to stderr, so both streams must be captured. If the +// stream is NOT framed (a raw TTY attach), it is returned unchanged. +func readDemuxedStream(reader io.Reader) (string, error) { + if reader == nil { + return "", nil } - return result.String() + raw, err := io.ReadAll(reader) + if err != nil { + return "", fmt.Errorf("reading container stream: %w", err) + } + + if !looksMultiplexed(raw) { + // Raw (TTY) attach: no framing to strip. + return string(raw), nil + } + + var out bytes.Buffer + // destOut and destErr both target out so stdout and stderr merge into one + // clean log buffer (act emits its JSON event stream on stderr). + if _, err := stdcopy.StdCopy(&out, &out, bytes.NewReader(raw)); err != nil { + return "", fmt.Errorf("demultiplexing container stream: %w", err) + } + return out.String(), nil +} + +// looksMultiplexed reports whether data begins with a well-formed Docker stream +// frame header: a known stream type, zeroed padding, and a length that does not +// overrun the buffer. This distinguishes a multiplexed (non-TTY) attach from a +// raw (TTY) attach so demuxing is only applied when there is framing to strip. +func looksMultiplexed(data []byte) bool { + if len(data) < dockerStreamHeaderLen { + return false + } + switch data[0] { + case byte(stdcopy.Stdin), byte(stdcopy.Stdout), byte(stdcopy.Stderr): + default: + return false + } + if data[1] != 0 || data[2] != 0 || data[3] != 0 { + return false + } + frameSize := binary.BigEndian.Uint32(data[4:dockerStreamHeaderLen]) + return uint64(dockerStreamHeaderLen)+uint64(frameSize) <= uint64(len(data)) } diff --git a/e2e/harness/act_test.go b/e2e/harness/act_test.go index 145f125b..16fc155c 100644 --- a/e2e/harness/act_test.go +++ b/e2e/harness/act_test.go @@ -1,14 +1,120 @@ package harness import ( + "bytes" "context" + "encoding/binary" + "strings" "testing" "time" + "github.com/moby/moby/api/pkg/stdcopy" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// frameDockerStream builds a single Docker hijacked-attach frame: an 8-byte +// header (1-byte stream type, 3 zero padding bytes, big-endian uint32 length) +// followed by the payload. This is the exact framing a TTY-less container exec +// produces on CI Linux, and the framing a raw io.Copy would leave interspersed +// in the captured logs. +func frameDockerStream(t *testing.T, stream stdcopy.StdType, payload string) []byte { + t.Helper() + var buf bytes.Buffer + header := make([]byte, dockerStreamHeaderLen) + header[0] = byte(stream) + binary.BigEndian.PutUint32(header[4:], uint32(len(payload))) + buf.Write(header) + buf.WriteString(payload) + return buf.Bytes() +} + +// TestReadDemuxedStream_MultiplexedFramesAreCleaned is the load-bearing proof +// for the CI-only act-output corruption: it feeds a hand-constructed MULTIPLEXED +// stream (stdout + stderr frames, each carrying Docker's 8-byte header) through +// readDemuxedStream and asserts the result is clean text with the streams merged +// and NO header bytes left behind. Before the demux fix the capture used a raw +// io.Copy, so these header bytes would remain interspersed and corrupt the JSON +// the act parser consumes. +func TestReadDemuxedStream_MultiplexedFramesAreCleaned(t *testing.T) { + t.Parallel() + + stdoutPayload := "hello world\n" + stderrPayload := `{"level":"info","msg":"Using docker host..."}` + "\n" + + var raw bytes.Buffer + raw.Write(frameDockerStream(t, stdcopy.Stdout, stdoutPayload)) + raw.Write(frameDockerStream(t, stdcopy.Stderr, stderrPayload)) + + got, err := readDemuxedStream(bytes.NewReader(raw.Bytes())) + require.NoError(t, err) + + // stdout and stderr are merged into a single clean log buffer. + assert.Contains(t, got, "hello world") + assert.Contains(t, got, `{"level":"info","msg":"Using docker host..."}`) + + // No Docker stream-header bytes survive. \x02\x00 is the stderr-frame + // header signature seen leading the corrupted CI logs; \x01\x00 is stdout. + assert.NotContains(t, got, "\x02\x00") + assert.NotContains(t, got, "\x01\x00") + assert.False(t, strings.ContainsRune(got, '\x00'), "no NUL padding bytes should remain") +} + +// TestReadDemuxedStream_RawStreamPassesThrough verifies a raw (TTY) attach with +// no Docker framing is returned unchanged, so the demux path never mangles an +// already-clean stream (the Docker Desktop / TTY case). +func TestReadDemuxedStream_RawStreamPassesThrough(t *testing.T) { + t.Parallel() + + raw := `{"level":"info","msg":"already clean"}` + "\nplain line\n" + got, err := readDemuxedStream(strings.NewReader(raw)) + require.NoError(t, err) + assert.Equal(t, raw, got) +} + +// TestReadDemuxedStream_NilReader confirms a nil reader yields empty output +// rather than panicking (the Exec call can return a nil reader). +func TestReadDemuxedStream_NilReader(t *testing.T) { + t.Parallel() + + got, err := readDemuxedStream(nil) + require.NoError(t, err) + assert.Empty(t, got) +} + +// TestLooksMultiplexed table-checks the frame-header detector that gates demux: +// a valid stdout/stderr frame is multiplexed; raw JSON, short input, bad +// padding, an unknown stream type, and an overrunning length are not. +func TestLooksMultiplexed(t *testing.T) { + t.Parallel() + + validFrame := frameDockerStream(t, stdcopy.Stdout, "hello world\n") + badPadding := append([]byte{1, 1, 0, 0, 0, 0, 0, 3}, []byte("abc")...) + unknownType := append([]byte{9, 0, 0, 0, 0, 0, 0, 3}, []byte("abc")...) + overrun := []byte{1, 0, 0, 0, 0, 0, 0, 255} + + tests := []struct { + name string + data []byte + want bool + }{ + {name: "valid stdout frame", data: validFrame, want: true}, + {name: "raw json is not framed", data: []byte(`{"level":"info"}`), want: false}, + {name: "shorter than header", data: []byte{1, 0, 0}, want: false}, + {name: "non-zero padding", data: badPadding, want: false}, + {name: "unknown stream type", data: unknownType, want: false}, + {name: "length overruns buffer", data: overrun, want: false}, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tt.want, looksMultiplexed(tt.data)) + }) + } +} + func TestActRunner_Start(t *testing.T) { if testing.Short() { t.Skip("skipping integration test")