Skip to content

Commit 26b9def

Browse files
blue-keleherrender-oss-copybara-sync[bot]
blue-keleher
authored and
render-oss-copybara-sync[bot]
committed
[CAP-8992] fix tasks dev hang when start command is invalid or crashes (#300)
StartService now monitors process exit via cmd.Wait() and returns a done channel. PopulateTasks and StartTask select on it to detect early exit instead of blocking forever on SDK channels that will never receive new data. Process crashes during task runs are marked as failed with a descriptive error message. GitOrigin-RevId: 157dfaf8a830c6cac875a882033a8284c10d3798
1 parent 3434360 commit 26b9def

3 files changed

Lines changed: 160 additions & 18 deletions

File tree

‎pkg/workflows/orchestrator/coordinator.go‎

Lines changed: 27 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ type Coordinator struct {
3838
}
3939

4040
type sdkExec interface {
41-
StartService(ctx context.Context, taskRunID string, socket string, mode Mode) (CleanupFunc, error)
41+
StartService(ctx context.Context, taskRunID string, socket string, mode Mode) (CleanupFunc, <-chan error, error)
4242
}
4343

4444
type serverFactory interface {
@@ -135,8 +135,8 @@ func (c *Coordinator) StartTask(ctx context.Context, taskIdentifier string, inpu
135135

136136
server.Start()
137137

138-
// Pass in a background context to avoid
139-
cleanupFunc, err := c.sdkExec.StartService(context.Background(), taskRun.ID, socket.Addr().String(), ModeRun)
138+
// Pass in a background context to avoid cancellation from the caller
139+
cleanupFunc, processDone, err := c.sdkExec.StartService(context.Background(), taskRun.ID, socket.Addr().String(), ModeRun)
140140
if err != nil {
141141
return nil, err
142142
}
@@ -146,11 +146,23 @@ func (c *Coordinator) StartTask(ctx context.Context, taskIdentifier string, inpu
146146
go func() {
147147
defer cleanupFunc()
148148

149-
callback := <-server.Channels.PostCallback
150-
151-
err := c.completeTask(callback.Body, taskRun.ID)
152-
if err != nil {
153-
fmt.Println("error completing task", err)
149+
select {
150+
case callback := <-server.Channels.PostCallback:
151+
err := c.completeTask(callback.Body, taskRun.ID)
152+
if err != nil {
153+
fmt.Println("error completing task", err)
154+
}
155+
case err := <-processDone:
156+
errMsg := "start command exited before completing the task"
157+
if err != nil {
158+
errMsg = fmt.Sprintf("%s: %s", errMsg, err)
159+
}
160+
updated, failErr := c.store.FailTaskRun(taskRun.ID, errMsg)
161+
if failErr != nil {
162+
fmt.Println("error failing task", failErr)
163+
return
164+
}
165+
c.statusReporter.TaskFailed(updated)
154166
}
155167
}()
156168

@@ -169,16 +181,21 @@ func (c *Coordinator) PopulateTasks(ctx context.Context) ([]*store.Task, error)
169181

170182
// we don't need to pass in a task run id here, because we don't need to
171183
// keep track of the logs for registration
172-
cleanupFunc, err := c.sdkExec.StartService(context.Background(), "", socket.Addr().String(), ModeRegister)
184+
cleanupFunc, processDone, err := c.sdkExec.StartService(context.Background(), "", socket.Addr().String(), ModeRegister)
173185
if err != nil {
174186
return nil, err
175187
}
176188
defer cleanupFunc()
177189

178-
// Wait until either the context is done or the server has received the tasks
190+
// Wait until either the context is done, the process exits, or the server has received the tasks
179191
select {
180192
case <-ctx.Done():
181193
return nil, ctx.Err()
194+
case err := <-processDone:
195+
if err != nil {
196+
return nil, fmt.Errorf("start command exited before registering tasks: %w\nRun with --debug to see process output", err)
197+
}
198+
return nil, fmt.Errorf("start command exited before registering tasks\nRun with --debug to see process output")
182199
case tasks := <-server.Channels.PostTasks:
183200
c.store.SetTasks(tasks.Body.Tasks)
184201
return c.store.GetTasks(), nil

‎pkg/workflows/orchestrator/coordinator_test.go‎

Lines changed: 122 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package orchestrator_test
33
import (
44
"context"
55
"encoding/json"
6+
"fmt"
67
"net"
78
"testing"
89
"time"
@@ -14,13 +15,18 @@ import (
1415
)
1516

1617
type fakeSdkExec struct {
17-
startService func(ctx context.Context, taskRunID string, socket string, mode orchestrator.Mode) (cleanupFunc func() error, err error)
18+
startService func(ctx context.Context, taskRunID string, socket string, mode orchestrator.Mode) (cleanupFunc func() error, processDone <-chan error, err error)
1819
}
1920

20-
func (f *fakeSdkExec) StartService(ctx context.Context, taskRunID string, socket string, mode orchestrator.Mode) (orchestrator.CleanupFunc, error) {
21+
func (f *fakeSdkExec) StartService(ctx context.Context, taskRunID string, socket string, mode orchestrator.Mode) (orchestrator.CleanupFunc, <-chan error, error) {
2122
return f.startService(ctx, taskRunID, socket, mode)
2223
}
2324

25+
// neverExits returns a process-done channel that never fires, simulating a long-running process.
26+
func neverExits() <-chan error {
27+
return make(chan error)
28+
}
29+
2430
type fakeServerFactory struct {
2531
newHandler func(socket net.Listener, input taskserver.GetInput200JSONResponse, getSubtaskResultFunc taskserver.GetSubtaskResultFunc, startSubtaskFunc taskserver.StartSubtaskFunc) *taskserver.ServerHandler
2632
}
@@ -75,15 +81,15 @@ func TestStartTask(t *testing.T) {
7581
ctx,
7682
s,
7783
&fakeSdkExec{
78-
startService: func(ctx context.Context, taskRunID string, socket string, mode orchestrator.Mode) (func() error, error) {
84+
startService: func(ctx context.Context, taskRunID string, socket string, mode orchestrator.Mode) (func() error, <-chan error, error) {
7985
if startCount == 0 {
8086
require.Equal(t, mode, orchestrator.ModeRegister)
8187
startCount++
8288
} else {
8389
require.Equal(t, mode, orchestrator.ModeRun)
8490
}
8591

86-
return cleanupFunc, nil
92+
return cleanupFunc, neverExits(), nil
8793
},
8894
},
8995
socketTracker,
@@ -162,8 +168,8 @@ func TestStartTaskWithSubtask(t *testing.T) {
162168
ctx,
163169
s,
164170
&fakeSdkExec{
165-
startService: func(ctx context.Context, taskRunID string, socket string, mode orchestrator.Mode) (func() error, error) {
166-
return cleanupFunc, nil
171+
startService: func(ctx context.Context, taskRunID string, socket string, mode orchestrator.Mode) (func() error, <-chan error, error) {
172+
return cleanupFunc, neverExits(), nil
167173
},
168174
},
169175
socketTracker,
@@ -244,3 +250,113 @@ func TestStartTaskWithSubtask(t *testing.T) {
244250
return taskRun.(taskserver.PostGetSubtaskResult200JSONResponse).Complete != nil
245251
}, time.Second*2, time.Millisecond*10)
246252
}
253+
254+
func TestPopulateTasksProcessExits(t *testing.T) {
255+
ctx := context.Background()
256+
257+
s := store.NewTaskStore()
258+
259+
postTasksChan := make(chan taskserver.PostRegisterTasksRequestObject)
260+
261+
socketTracker, err := orchestrator.NewSocketTracker(ctx)
262+
require.NoError(t, err)
263+
264+
coordinator := orchestrator.NewCoordinator(
265+
ctx,
266+
s,
267+
&fakeSdkExec{
268+
startService: func(ctx context.Context, taskRunID string, socket string, mode orchestrator.Mode) (func() error, <-chan error, error) {
269+
processDone := make(chan error, 1)
270+
processDone <- fmt.Errorf("exit status 1")
271+
return func() error { return nil }, processDone, nil
272+
},
273+
},
274+
socketTracker,
275+
&fakeServerFactory{
276+
newHandler: func(socket net.Listener, input taskserver.GetInput200JSONResponse, getSubtaskResultFunc taskserver.GetSubtaskResultFunc, startSubtaskFunc taskserver.StartSubtaskFunc) *taskserver.ServerHandler {
277+
return &taskserver.ServerHandler{
278+
Socket: socket,
279+
Input: input,
280+
Channels: taskserver.ServerChannels{
281+
PostCallback: make(chan taskserver.PostCallbackRequestObject),
282+
PostTasks: postTasksChan,
283+
},
284+
}
285+
},
286+
},
287+
&noopStatusReporter{},
288+
)
289+
290+
_, err = coordinator.PopulateTasks(ctx)
291+
require.Error(t, err)
292+
require.Contains(t, err.Error(), "start command exited before registering tasks")
293+
require.Contains(t, err.Error(), "exit status 1")
294+
require.Contains(t, err.Error(), "--debug")
295+
}
296+
297+
func TestStartTaskProcessExits(t *testing.T) {
298+
ctx := context.Background()
299+
300+
s := store.NewTaskStore()
301+
302+
tasks := []taskserver.Task{
303+
{
304+
Name: "test-task",
305+
},
306+
}
307+
308+
postTasksChan := make(chan taskserver.PostRegisterTasksRequestObject, 1)
309+
postTasksChan <- taskserver.PostRegisterTasksRequestObject{
310+
Body: &taskserver.PostRegisterTasksJSONRequestBody{
311+
Tasks: tasks,
312+
},
313+
}
314+
315+
socketTracker, err := orchestrator.NewSocketTracker(ctx)
316+
require.NoError(t, err)
317+
318+
startCount := 0
319+
coordinator := orchestrator.NewCoordinator(
320+
ctx,
321+
s,
322+
&fakeSdkExec{
323+
startService: func(ctx context.Context, taskRunID string, socket string, mode orchestrator.Mode) (func() error, <-chan error, error) {
324+
startCount++
325+
if startCount == 1 {
326+
// Registration succeeds normally
327+
return func() error { return nil }, neverExits(), nil
328+
}
329+
// Task run: process exits immediately
330+
processDone := make(chan error, 1)
331+
processDone <- fmt.Errorf("exit status 1")
332+
return func() error { return nil }, processDone, nil
333+
},
334+
},
335+
socketTracker,
336+
&fakeServerFactory{
337+
newHandler: func(socket net.Listener, input taskserver.GetInput200JSONResponse, getSubtaskResultFunc taskserver.GetSubtaskResultFunc, startSubtaskFunc taskserver.StartSubtaskFunc) *taskserver.ServerHandler {
338+
return &taskserver.ServerHandler{
339+
Socket: socket,
340+
Input: input,
341+
Channels: taskserver.ServerChannels{
342+
PostCallback: make(chan taskserver.PostCallbackRequestObject),
343+
PostTasks: postTasksChan,
344+
},
345+
}
346+
},
347+
},
348+
&noopStatusReporter{},
349+
)
350+
351+
taskRun, err := coordinator.StartTask(ctx, "fake-workflow/test-task", []byte{}, nil)
352+
require.NoError(t, err)
353+
354+
// The background goroutine should detect the process exit and fail the task run
355+
require.Eventually(t, func() bool {
356+
tr := s.GetTaskRun(taskRun.ID)
357+
return tr.Status == store.TaskRunStatusFailed
358+
}, time.Second*2, time.Millisecond*10)
359+
360+
tr := s.GetTaskRun(taskRun.ID)
361+
require.Contains(t, *tr.Error, "start command exited before completing the task")
362+
}

‎pkg/workflows/orchestrator/exec.go‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@ func NewExec(logsStore *logs.LogStore, debug bool, command string, args ...strin
4141
}
4242
}
4343

44-
func (e *Exec) StartService(ctx context.Context, taskRunID string, socketPath string, mode Mode) (CleanupFunc, error) {
44+
func (e *Exec) StartService(ctx context.Context, taskRunID string, socketPath string, mode Mode) (CleanupFunc, <-chan error, error) {
4545
cmd := exec.CommandContext(ctx, e.command, e.args...)
4646
cmd.Env = append(os.Environ(), fmt.Sprintf(SocketPathEnv+"=%s", socketPath), fmt.Sprintf(ModeEnv+"=%s", mode))
4747

@@ -64,7 +64,16 @@ func (e *Exec) StartService(ctx context.Context, taskRunID string, socketPath st
6464
pt.Kill()
6565
}()
6666

67+
if err := cmd.Start(); err != nil {
68+
return nil, nil, err
69+
}
70+
71+
processDone := make(chan error, 1)
72+
go func() {
73+
processDone <- cmd.Wait()
74+
}()
75+
6776
return func() error {
6877
return pt.Kill()
69-
}, cmd.Start()
78+
}, processDone, nil
7079
}

0 commit comments

Comments
 (0)