Merge branch 'pr/450'

Author: F1bonacc1

src/app/process.go

@@ -110,7 +110,18 @@ func NewProcess(opts ...ProcOpts) *Process {
 
 func (p *Process) run() int {
 	if p.isState(types.ProcessStateTerminating) {
-		return 0
+		select {
+		case <-p.procRunCtx.Done():
+			// Concurrent stop in progress — honour it.
+			return 0
+		default:
+			// Stale "Terminating" state from a prior broken shutdown (e.g. zombie
+			// that was reaped without completing the normal lifecycle). Reset so
+			// the process can start fresh.
+			log.Warn().Str("process", p.getName()).
+				Msg("Resetting stale Terminating state before start")
+			p.setState(types.ProcessStatePending)
+		}
 	}
 
 	if err := p.validateProcess(); err != nil {
@@ -187,13 +198,24 @@ func (p *Process) waitForStdOutErr() {
 	ctx, cancel := context.WithCancel(context.Background())
 	if p.procConf.IsDaemon {
 		ctx, cancel = context.WithTimeout(context.Background(), time.Duration(p.procConf.LaunchTimeout)*time.Second)
+	} else if p.isState(types.ProcessStateTerminating) {
+		// In terminating state, never block forever on stdout/stderr drain.
+		// Reaching command.Wait() is required to reap child processes.
+		timeoutSec := p.procConf.ShutDownParams.ShutDownTimeout
+		if timeoutSec == UndefinedShutdownTimeoutSec {
+			timeoutSec = DefaultShutdownTimeoutSec
+		}
+		ctx, cancel = context.WithTimeout(context.Background(), time.Duration(timeoutSec)*time.Second)
 	}
 	defer cancel()
 	if p.stdOutDone != nil {
 		select {
 		case <-ctx.Done():
 			log.Debug().Msgf("%s stdout done with timeout", p.getName())
 			return
+		case <-p.procRunCtx.Done():
+			log.Debug().Msgf("%s stdout abandoned: process stopped", p.getName())
+			return
 		case <-p.stdOutDone:
 			log.Debug().Msgf("%s stdout done", p.getName())
 		}
@@ -204,6 +226,9 @@ func (p *Process) waitForStdOutErr() {
 		case <-ctx.Done():
 			log.Debug().Msgf("%s stderr done with timeout", p.getName())
 			return
+		case <-p.procRunCtx.Done():
+			log.Debug().Msgf("%s stderr abandoned: process stopped", p.getName())
+			return
 		case <-p.stdErrDone:
 			log.Debug().Msgf("%s stderr done", p.getName())
 		}

src/app/system_test.go

@@ -1431,6 +1431,157 @@ func TestSystem_ConcurrentRestartRaceCondition(t *testing.T) {
 	}
 }
 
+func TestSystem_StartProcessResetsStaleTerminatingState(t *testing.T) {
+	testProcess := "stale_terminating"
+	shell := command.DefaultShellConfig()
+
+	project := &types.Project{
+		Processes: map[string]types.ProcessConfig{
+			testProcess: {
+				Name:        testProcess,
+				ReplicaName: testProcess,
+				Executable:  shell.ShellCommand,
+				Args:        []string{shell.ShellArgument, getSleepCommand(1.0)},
+				RestartPolicy: types.RestartPolicyConfig{
+					Restart: types.RestartPolicyNo,
+				},
+			},
+		},
+		ShellConfig: shell,
+	}
+
+	runner, err := NewProjectRunner(&ProjectOpts{
+		project:         project,
+		processesToRun:  []string{},
+		noDeps:          false,
+		mainProcess:     "",
+		mainProcessArgs: []string{},
+		isTuiOn:         false,
+	})
+	if err != nil {
+		t.Error(err.Error())
+		return
+	}
+
+	// Simulate stale state from a prior broken termination.
+	runner.statesMutex.Lock()
+	runner.processStates[testProcess].Status = types.ProcessStateTerminating
+	runner.processStates[testProcess].IsRunning = false
+	runner.statesMutex.Unlock()
+	runner.runProcMutex.Lock()
+	runner.runningProcesses = make(map[string]*Process)
+	runner.runProcMutex.Unlock()
+	runner.doneProcMutex.Lock()
+	runner.doneProcesses = make(map[string]*Process)
+	runner.doneProcMutex.Unlock()
+	runner.logger = pclog.NewNilLogger()
+
+	if err := runner.StartProcess(testProcess); err != nil {
+		t.Fatalf("failed to start process: %v", err)
+	}
+
+	var lastStatus string
+	for attempts := range 200 {
+		state, stateErr := runner.GetProcessState(testProcess)
+		if stateErr != nil {
+			t.Fatalf("failed to get process state: %v", stateErr)
+		}
+		lastStatus = state.Status
+		if state.Status == types.ProcessStateRunning || state.Status == types.ProcessStateCompleted {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+		if attempts == 199 {
+			t.Fatalf("process failed to leave stale Terminating state, last status=%s", state.Status)
+		}
+	}
+
+	if lastStatus == types.ProcessStateTerminating {
+		t.Fatalf("expected process to recover from stale Terminating state, got %s", lastStatus)
+	}
+
+	// Cleanup if still running.
+	if runner.getRunningProcess(testProcess) != nil {
+		if err := runner.StopProcess(testProcess); err != nil {
+			t.Fatalf("failed to stop process: %v", err)
+		}
+	}
+}
+
+func TestSystem_TestTerminatingWithOrphanedChildHoldingPipe(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("test requires Unix process semantics")
+	}
+	// Reproduce the zombie/stale-Terminating scenario:
+	// A parent process spawns a child that inherits stdout and keeps it open.
+	// When we stop the parent, the stdout pipe never gets an EOF because the
+	// child still holds it. Without a timeout in waitForStdOutErr, the
+	// process would hang in Terminating forever and never reach Completed.
+
+	testProcess := "orphan_pipe"
+	shell := command.DefaultShellConfig()
+	shutdownTimeout := 3
+
+	project := &types.Project{
+		Processes: map[string]types.ProcessConfig{
+			testProcess: {
+				Name:        testProcess,
+				ReplicaName: testProcess,
+				Executable:  shell.ShellCommand,
+				// Parent spawns a child in a new session (setsid) so it won't
+				// receive the SIGTERM sent to the parent's process group.
+				// The child writes to stdout (holding the pipe open) and sleeps.
+				// When the parent is killed, the child survives and keeps the
+				// pipe open, blocking waitForStdOutErr until the timeout fires.
+				Args: []string{shell.ShellArgument, "setsid bash -c 'while true; do echo holding_pipe; sleep 1; done' & sleep 60"},
+				RestartPolicy: types.RestartPolicyConfig{
+					Restart: types.RestartPolicyNo,
+				},
+				ShutDownParams: types.ShutDownParams{
+					ShutDownTimeout: shutdownTimeout,
+					Signal:          int(syscall.SIGTERM),
+				},
+			},
+		},
+		ShellConfig: shell,
+	}
+
+	runner, err := NewProjectRunner(&ProjectOpts{project: project})
+	if err != nil {
+		t.Fatal(err)
+	}
+	go func() {
+		_ = runner.Run()
+	}()
+
+	// Wait for the process to be running (poll via mutex-protected method to avoid race).
+	var proc *Process
+	for i := 0; i < 50; i++ {
+		time.Sleep(100 * time.Millisecond)
+		proc = runner.getRunningProcess(testProcess)
+		if proc != nil {
+			break
+		}
+	}
+	if proc == nil {
+		t.Fatal("process never started")
+	}
+
+	// Stop the parent — it will enter Terminating, but the child keeps stdout open.
+	stopDone := make(chan struct{})
+	go func() {
+		defer close(stopDone)
+		_ = runner.StopProcess(testProcess)
+	}()
+
+	// The process must reach Completed within the shutdown timeout + margin,
+	// NOT stay stuck in Terminating.
+	waitForProcessState(t, runner, testProcess, types.ProcessStateCompleted,
+		time.Duration(shutdownTimeout+5)*time.Second)
+
+	<-stopDone
+}
+
 func TestReadinessProbeRestart(t *testing.T) {
 	proc := &types.ProcessConfig{
 		Name:        "test",