[ACI-4154] Log interceptor potential deadlock fix (#286)

* Resolve deadlock by write calls with independent timeouts

* Cleanup comments, imports

* Cancel context of writes sooner

* Reduce expectations in case of slow writes

* Use input timeout in both writer calls

* Make more concise non blocking writer

* Use byte array in NonBlockingWriter

* Change the write method to match change in channel type

* Make buffer full errors debug only and raise buffer size

Author: Badlazzor

autocodesign/devportalclient/appstoreconnect/appstoreconnect.go

@@ -283,7 +283,6 @@ func (c *Client) Do(req *http.Request, v interface{}) (*http.Response, error) {
 
 	c.tracker.TrackAPIRequest(req.Method, req.URL.Host, req.URL.Path, resp.StatusCode, duration)
 
-
 	if v != nil {
 		decErr := json.NewDecoder(resp.Body).Decode(v)
 		if decErr == io.EOF {

destination/device_finder.go

@@ -4,11 +4,12 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/bitrise-io/go-utils/errorutil"
-	"github.com/bitrise-io/go-utils/retry"
 	"os"
 	"time"
 
+	"github.com/bitrise-io/go-utils/errorutil"
+	"github.com/bitrise-io/go-utils/retry"
+
 	"github.com/bitrise-io/go-utils/v2/command"
 	"github.com/bitrise-io/go-utils/v2/log"
 	"github.com/bitrise-io/go-xcode/v2/xcodeversion"

destination/parse.go

@@ -2,11 +2,12 @@ package destination
 
 import (
 	"fmt"
+	"os"
+	"strings"
+
 	"github.com/bitrise-io/go-utils/errorutil"
 	"github.com/bitrise-io/go-utils/v2/command"
 	"github.com/hashicorp/go-version"
-	"os"
-	"strings"
 )
 
 // DeviceList ...

loginterceptor/loginterceptor.go

@@ -13,8 +13,8 @@ import (
 // both writers. Partial writes without newline are buffered until a newline.
 type PrefixInterceptor struct {
 	prefixRegexp *regexp.Regexp
-	intercepted  io.Writer
-	target       io.Writer
+	intercepted  *NonBlockingWriter
+	target       *NonBlockingWriter
 	logger       log.Logger
 
 	// internal pipe and goroutine to scan and route
@@ -31,8 +31,8 @@ func NewPrefixInterceptor(prefixRegexp *regexp.Regexp, intercepted, target io.Wr
 	pipeReader, pipeWriter := io.Pipe()
 	interceptor := &PrefixInterceptor{
 		prefixRegexp:   prefixRegexp,
-		intercepted:    intercepted,
-		target:         target,
+		intercepted:    NewNonBlockingWriter(intercepted, logger),
+		target:         NewNonBlockingWriter(target, logger),
 		logger:         logger,
 		internalReader: pipeReader,
 		internalWriter: pipeWriter,
@@ -55,18 +55,12 @@ func (i *PrefixInterceptor) Close() error {
 }
 
 func (i *PrefixInterceptor) closeAfterRun() {
-	// Close writers if able
-	if interceptedCloser, ok := i.intercepted.(io.Closer); ok {
-		if err := interceptedCloser.Close(); err != nil {
-			i.logger.Errorf("closing intercepted writer: %v", err)
-		}
+	if err := i.intercepted.Close(); err != nil {
+		i.logger.Errorf("intercepted writer: %v", err)
 	}
-	if originalCloser, ok := i.target.(io.Closer); ok {
-		if err := originalCloser.Close(); err != nil {
-			i.logger.Errorf("closing original writer: %v", err)
-		}
+	if err := i.target.Close(); err != nil {
+		i.logger.Errorf("target writer: %v", err)
 	}
-
 	if err := i.internalReader.Close(); err != nil {
 		i.logger.Errorf("internal reader: %v", err)
 	}
@@ -104,3 +98,53 @@ func (i *PrefixInterceptor) run() {
 		i.logger.Errorf("router scanner error: %v\n", err)
 	}
 }
+
+// NonBlockingWriter is an io.Writer that writes to a wrapped io.Writer in a non-blocking way.
+type NonBlockingWriter struct {
+	channel chan []byte
+	wrapped io.Writer
+	logger  log.Logger
+}
+
+// NewNonBlockingWriter creates a new NonBlockingWriter.
+func NewNonBlockingWriter(w io.Writer, logger log.Logger) *NonBlockingWriter {
+	writer := &NonBlockingWriter{
+		channel: make(chan []byte, 10000), // buffered channel to avoid blocking
+		wrapped: w,
+		logger:  logger,
+	}
+	go writer.Run()
+	return writer
+}
+
+// Write implements io.Writer. It writes into an internal pipe which the interceptor goroutine consumes.
+func (i *NonBlockingWriter) Write(p []byte) (int, error) {
+	select {
+	case i.channel <- p:
+		return len(p), nil
+	default:
+		i.logger.Debugf("buffer full, dropping log")
+		return 0, nil
+	}
+}
+
+// Close stops the interceptor and closes the pipe.
+func (i *NonBlockingWriter) Close() error {
+	close(i.channel)
+	return nil
+}
+
+// Run consumes the channel and writes to the wrapped writer.
+func (i *NonBlockingWriter) Run() {
+	for msg := range i.channel {
+		if _, err := i.wrapped.Write(msg); err != nil {
+			i.logger.Errorf("NonBlockingWriter: wrapped writer error: %v", err)
+		}
+	}
+
+	if closer, ok := i.wrapped.(io.Closer); ok {
+		if err := closer.Close(); err != nil {
+			i.logger.Errorf("NonBlockingWriter: closing wrapped writer: %v", err)
+		}
+	}
+}

loginterceptor/loginterceptor_test.go

@@ -40,6 +40,64 @@ func TestPrefixInterceptor(t *testing.T) {
 	assert.Equal(t, msg1+msg2+msg3+msg4, string(target))
 }
 
+func TestPrefixInterceptorWithPrematureClose(t *testing.T) {
+	interceptReader, interceptWriter := io.Pipe()
+	targetReader, targetWriter := io.Pipe()
+	re := regexp.MustCompile(`^\[Bitrise.*\].*`)
+
+	sut := loginterceptor.NewPrefixInterceptor(re, interceptWriter, targetWriter, log.NewLogger())
+
+	msg1 := "Log message without prefix\n"
+	msg2 := "[Bitrise Analytics] Log message with prefix\n"
+	msg3 := "[Bitrise Build Cache] Log message with prefix\n"
+	msg4 := "Last message that won't be sent\n"
+
+	go func() {
+		_, _ = sut.Write([]byte(msg1))
+		_, _ = sut.Write([]byte(msg2))
+		_, _ = sut.Write([]byte(msg3))
+		_ = sut.Close()
+		_, _ = sut.Write([]byte(msg4))
+	}()
+
+	intercepted, target, err := readTwo(interceptReader, targetReader)
+	assert.NoError(t, err)
+	assert.Equal(t, msg2+msg3, string(intercepted))
+	assert.Equal(t, msg1+msg2+msg3, string(target))
+}
+
+func TestPrefixInterceptorWithBlockedPipe(t *testing.T) {
+	interceptReader, interceptWriter := io.Pipe()
+	targetReader, targetWriter := io.Pipe()
+	re := regexp.MustCompile(`^\[Bitrise.*\].*`)
+
+	sut := loginterceptor.NewPrefixInterceptor(re, interceptWriter, targetWriter, log.NewLogger())
+
+	msg1 := "Log message without prefix\n"
+	msg2 := "[Bitrise Analytics] Log message with prefix\n"
+	msg3 := "[Bitrise Build Cache] Log message with prefix\n"
+	msg4 := "Stuff [Bitrise Build Cache] Log message without prefix\n"
+
+	go func() {
+		//nolint:errCheck
+		defer sut.Close()
+
+		_, _ = sut.Write([]byte(msg1))
+		_, _ = sut.Write([]byte(msg2))
+		_, _ = sut.Write([]byte(msg3))
+		_, _ = sut.Write([]byte(msg4))
+	}()
+
+	target, err := io.ReadAll(targetReader)
+	assert.NoError(t, err)
+	assert.Equal(t, msg1+msg2+msg3+msg4, string(target))
+
+	// Reading from interceptReader should block until targetWriter is read
+	intercepted, err := io.ReadAll(interceptReader)
+	assert.NoError(t, err)
+	assert.Equal(t, msg2+msg3, string(intercepted))
+}
+
 func readTwo(r1, r2 io.Reader) (out1, out2 []byte, err error) {
 	var (
 		wg     sync.WaitGroup

xcodecache/swiftpm_cache.go

@@ -12,7 +12,8 @@ import (
 // SwiftPackagesStateInvalid is the partial error message printed out if swift packages cache is invalid.
 // Can be used to detect invalid state and clear the path returned by SwiftPackagesPath.
 // xcodebuild: error: Could not resolve package dependencies:
-//   The repository at [path] is invalid; try resetting package caches
+//
+//	The repository at [path] is invalid; try resetting package caches
 const SwiftPackagesStateInvalid = "Could not resolve package dependencies:"
 
 // SwiftPackageCache ...