Atp concurrency improvements

atp/server.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"os"
 	"sync"
+	"time"
 )
 
 // RunATPServer runs an ArcaflowTransportProtocol server with a given schema.
@@ -83,12 +84,23 @@ func initializeATPServerSession(
 
 func (s *atpServerSession) sendRuntimeMessage(msgID uint32, runID string, message any) error {
 	s.encoderMutex.Lock()
-	defer s.encoderMutex.Unlock()
-	return s.cborStdout.Encode(RuntimeMessage{
-		MessageID:   msgID,
-		RunID:       runID,
-		MessageData: message,
-	})
+	doneChannel := make(chan error, 1)
+	go func() {
+		defer close(doneChannel)
+		doneChannel <- s.cborStdout.Encode(RuntimeMessage{
+			MessageID:   msgID,
+			RunID:       runID,
+			MessageData: message,
+		})
+	}()
+	select {
+	case err := <-doneChannel:
+		s.encoderMutex.Unlock()
+		return err
+	case <-time.After(time.Second * 60):
+		s.encoderMutex.Unlock()
+		return fmt.Errorf("send timeout exceeded while sending message ID %q for run id %q", msgID, runID)
+	}
 }
 
 func (s *atpServerSession) handleClosure() []*ServerError {