From 4eea61d8263f6a8d4920e48c7917102f63df3fdf Mon Sep 17 00:00:00 2001
From: dawn <dawn@tangled.org>
Date: Fri, 18 Sep 2026 14:13:45 +0300
Subject: [PATCH] spindle/engines/nixery: bound the exec attach read with the
 workflow context

Signed-off-by: dawn <dawn@tangled.org>
---
 spindle/engines/nixery/engine.go      | 28 ++++++++++--
 spindle/engines/nixery/engine_test.go | 66 +++++++++++++++++++++++++++
 2 files changed, 91 insertions(+), 3 deletions(-)

diff --git a/spindle/engines/nixery/engine.go b/spindle/engines/nixery/engine.go
index 4f1fe79fd..248706984 100644
--- a/spindle/engines/nixery/engine.go
+++ b/spindle/engines/nixery/engine.go
@@ -34,6 +34,8 @@ const (
 	// how long a workflow's container gets to stop after the workflow ends,
 	// docker's own stop grace included
 	teardownTimeout = 30 * time.Second
+	// how long the reader gets to notice the stream was closed
+	closeGrace = 5 * time.Second
 )
 
 type cleanupFunc func(context.Context) error
@@ -406,9 +408,7 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m
 	}
 	defer execResp.Close()
 
-	// This is apparently best way to wait for the command to complete.
-	_, err = io.ReadAll(execResp.Reader)
-	if err != nil {
+	if err := awaitExec(ctx, execResp.Reader, execResp.Close); err != nil {
 		return err
 	}
 
@@ -430,6 +430,28 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m
 	return nil
 }
 
+// the exec stream outlives the workflow context, so a canceled or timed out
+// workflow has to close it or setup blocks forever holding its seat
+func awaitExec(ctx context.Context, stream io.Reader, closeStream func()) error {
+	done := make(chan error, 1)
+	go func() {
+		_, err := io.ReadAll(stream)
+		done <- err
+	}()
+
+	select {
+	case err := <-done:
+		return err
+	case <-ctx.Done():
+		closeStream()
+		select {
+		case <-done:
+		case <-time.After(closeGrace):
+		}
+		return ctx.Err()
+	}
+}
+
 func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
 	addl := w.Data.(addlFields)
 	workflowEnvs := ConstructEnvs(w.Environment)
diff --git a/spindle/engines/nixery/engine_test.go b/spindle/engines/nixery/engine_test.go
index 37b293ae3..5e024334c 100644
--- a/spindle/engines/nixery/engine_test.go
+++ b/spindle/engines/nixery/engine_test.go
@@ -2,6 +2,8 @@ package nixery
 
 import (
 	"context"
+	"errors"
+	"io"
 	"testing"
 	"time"
 
@@ -54,3 +56,67 @@ func TestDestroyWorkflowCleansUpAfterCancellation(t *testing.T) {
 		t.Fatal("cleanup did not run")
 	}
 }
+
+// only ends when closed, like a hijacked exec attach nothing will write to
+type hangingStream struct{ closed chan struct{} }
+
+func newHangingStream() *hangingStream { return &hangingStream{closed: make(chan struct{})} }
+
+func (s *hangingStream) Read([]byte) (int, error) {
+	<-s.closed
+	return 0, io.EOF
+}
+
+func (s *hangingStream) close() {
+	select {
+	case <-s.closed:
+	default:
+		close(s.closed)
+	}
+}
+
+type failingStream struct{ err error }
+
+func (s failingStream) Read([]byte) (int, error) { return 0, s.err }
+
+func TestAwaitExecUnblocksWhenTheWorkflowContextEnds(t *testing.T) {
+	stream := newHangingStream()
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	done := make(chan error, 1)
+	go func() { done <- awaitExec(ctx, stream, stream.close) }()
+
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("awaitExec() = %v, want context.Canceled", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("a cancelled workflow is still waiting on the exec stream")
+	}
+}
+
+func TestAwaitExecReturnsWhenTheStreamEnds(t *testing.T) {
+	stream := newHangingStream()
+
+	done := make(chan error, 1)
+	go func() { done <- awaitExec(context.Background(), stream, stream.close) }()
+	stream.close()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("awaitExec() = %v, want nil", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("awaitExec did not return after the stream ended")
+	}
+}
+
+func TestAwaitExecReturnsStreamErrors(t *testing.T) {
+	broken := errors.New("stream broke")
+	if err := awaitExec(context.Background(), failingStream{err: broken}, func() {}); !errors.Is(err, broken) {
+		t.Fatalf("awaitExec() = %v, want %v", err, broken)
+	}
+}
-- 
2.51.2