Skip to content

Commit 3339e02

Browse files
ostermanclaude
andcommitted
fix(output): serialize whole flush, not per-line, in LinePrefixWriter
writeLine acquired/released the shared writeMu once per line, so a single Write() call that produced multiple lines (e.g. a hook's buffered "\r"-then-"\n" progress update) could have another node's writer interleave a line in between, corrupting concurrent Terraform node output. Hold writeMu for the entire flush instead. Reproduced with `go test -race -count=200` on TestExecuteTerraformConcurrentHooksUseNodeWriters, which was flaking in CI (Acceptance Tests macos, job 92489708710); now passes 200/200. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1 parent 0200c64 commit 3339e02

1 file changed

Lines changed: 13 additions & 3 deletions

File tree

pkg/io/line_prefix_writer.go

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,13 @@ func (w *LinePrefixWriter) Write(p []byte) (int, error) {
6565
}
6666

6767
w.buffer = append(w.buffer, p...)
68+
69+
// Hold writeMu for the whole flush so that every line produced by this
70+
// single Write call reaches the shared writer as one contiguous block.
71+
// Locking per-line let a concurrent node's writer interleave a line in
72+
// between two lines emitted from the same Write call.
73+
w.writeMu.Lock()
74+
defer w.writeMu.Unlock()
6875
if err := w.flushCompleteLinesLocked(); err != nil {
6976
return 0, err
7077
}
@@ -81,6 +88,10 @@ func (w *LinePrefixWriter) Flush() error {
8188
if len(w.buffer) == 0 {
8289
return nil
8390
}
91+
92+
w.writeMu.Lock()
93+
defer w.writeMu.Unlock()
94+
8495
if err := w.flushCompleteLinesLocked(); err != nil {
8596
return err
8697
}
@@ -95,7 +106,7 @@ func (w *LinePrefixWriter) Flush() error {
95106
return nil
96107
}
97108

98-
// flushCompleteLinesLocked writes buffered complete lines while w.mu is held.
109+
// flushCompleteLinesLocked writes buffered complete lines while w.mu and w.writeMu are held.
99110
func (w *LinePrefixWriter) flushCompleteLinesLocked() error {
100111
for {
101112
idx := lineEndIndex(w.buffer)
@@ -115,12 +126,11 @@ func (w *LinePrefixWriter) flushCompleteLinesLocked() error {
115126
}
116127

117128
// writeLine writes one already-delimited line with the configured prefix.
129+
// Callers must hold w.writeMu.
118130
func (w *LinePrefixWriter) writeLine(line []byte) error {
119131
if w.w == nil {
120132
return nil
121133
}
122-
w.writeMu.Lock()
123-
defer w.writeMu.Unlock()
124134

125135
line = bytes.ReplaceAll(line, crlfBytes, lfBytes)
126136
line = bytes.ReplaceAll(line, crBytes, lfBytes)

0 commit comments

Comments
 (0)