diff --git a/internal/client/agent/executor.go b/internal/client/agent/executor.go index d680a2b..40eff2e 100644 --- a/internal/client/agent/executor.go +++ b/internal/client/agent/executor.go @@ -118,12 +118,10 @@ func (executor *Executor) launch(ctx context.Context, issue domain.UUID, revisio return err } // The native Windows supervisor creates/assigns the Job while the child is - // suspended and releases it before returning. Marking authorization before - // that call makes a daemon crash in any of those windows recover as an - // interrupted, non-redispatchable command. - if err := executor.Store.SetLaunchPhase(ctx, issue, domain.LaunchPhaseAuthorized, "pending", 0); err != nil { - return err - } + // suspended. It does not cross the durable authorization barrier until the + // process object has been fully prepared and its immutable identity is + // available. A daemon crash before this write leaves only launch_prepared; + // the kill-on-close Job prevents a suspended child from surviving restart. runContext, cancel := context.WithCancel(context.Background()) process, err := executor.Supervisor.Start(runContext, supervisor.StartSpec{IssueUUID: issue, CommandRevision: revision, Execution: proto.Clone(spec).(*rvboxv1.ExecutionSpec), ScriptBody: scriptBody, WorkingDirectory: workingDirectory, Environment: cloneEnvironment(spec.GetEnvOverrides()), ExecutionProfiles: executionProfileNames(spec.GetExecutionProfiles())}) if err != nil { @@ -141,6 +139,11 @@ func (executor *Executor) launch(ctx context.Context, issue domain.UUID, revisio cancel() return err } + if err := process.Release(runContext); err != nil { + _, _ = executor.Supervisor.Signal(context.Background(), process, supervisor.SignalKill) + cancel() + return err + } executor.mu.Lock() executor.active[issue] = process executor.cancel[issue] = cancel diff --git a/internal/client/supervisor/supervisor.go b/internal/client/supervisor/supervisor.go index e592bbc..407f4e4 100644 --- a/internal/client/supervisor/supervisor.go +++ b/internal/client/supervisor/supervisor.go @@ -130,6 +130,9 @@ func windowsExecutionContext(value string) (rvboxv1.WindowsExecutionContext, boo type Process interface { IssueUUID() domain.UUID Identity() EffectiveIdentity + // Release crosses the durable launch-authorized barrier. A native Windows + // process is created suspended and must not execute before this call. + Release(context.Context) error // ReadOutput returns the next bounded stdout/stderr chunk. It continues // until both child pipes reach EOF, so Wait never reports a terminal // result before the captured output has drained. diff --git a/internal/client/supervisor/windows/native_exec.go b/internal/client/supervisor/windows/native_exec.go index 9046254..ffb15c2 100644 --- a/internal/client/supervisor/windows/native_exec.go +++ b/internal/client/supervisor/windows/native_exec.go @@ -95,13 +95,16 @@ type execProcess struct { started time.Time waitFn func() (int32, bool, error) killFn func(uint32) error + releaseFn func() error snapshotFn func() (supervisor.ResourceSnapshot, error) - mu sync.Mutex - finished bool - status supervisor.ExitStatus - waitErr error - closeIn sync.Once + mu sync.Mutex + finished bool + status supervisor.ExitStatus + waitErr error + closeIn sync.Once + release sync.Once + releaseErr error } type outputResult struct { @@ -113,6 +116,21 @@ func (process *execProcess) IssueUUID() domain.UUID { return process.issue } func (process *execProcess) Identity() supervisor.EffectiveIdentity { return process.identity } +// Release is the second half of the durable launch barrier. The portable +// adapter has no suspended native handle, so its release is intentionally a +// no-op; the Windows adapter supplies a ResumeThread closure. +func (process *execProcess) Release(_ context.Context) error { + if process == nil { + return ErrProcessNotReady + } + process.release.Do(func() { + if process.releaseFn != nil { + process.releaseErr = process.releaseFn() + } + }) + return process.releaseErr +} + func (process *execProcess) ReadOutput(ctx context.Context) (supervisor.OutputChunk, error) { if process == nil || process.outputs == nil { return supervisor.OutputChunk{}, ErrProcessNotReady @@ -453,15 +471,15 @@ func (manager *execSupervisor) startCommand(ctx context.Context, spec supervisor } return code, false, err } - return manager.registerProcess(spec.IssueUUID, identity, command, stdin, stdout, stderr, started, waitFn, func(uint32) error { return command.Process.Kill() }, cleanup), nil + return manager.registerProcess(spec.IssueUUID, identity, command, stdin, stdout, stderr, started, waitFn, func(uint32) error { return command.Process.Kill() }, nil, cleanup), nil } -func (manager *execSupervisor) registerProcess(issue domain.UUID, identity supervisor.EffectiveIdentity, command *exec.Cmd, stdin io.WriteCloser, stdout, stderr io.ReadCloser, started time.Time, waitFn func() (int32, bool, error), killFn func(uint32) error, cleanup func()) *execProcess { +func (manager *execSupervisor) registerProcess(issue domain.UUID, identity supervisor.EffectiveIdentity, command *exec.Cmd, stdin io.WriteCloser, stdout, stderr io.ReadCloser, started time.Time, waitFn func() (int32, bool, error), killFn func(uint32) error, releaseFn func() error, cleanup func()) *execProcess { var pid uint32 if command != nil && command.Process != nil && command.Process.Pid > 0 { pid = uint32(command.Process.Pid) } - process := &execProcess{issue: issue, identity: identity, cmd: command, pid: pid, stdin: stdin, stdout: stdout, stderr: stderr, outputs: make(chan outputResult, 32), done: make(chan struct{}), started: started, waitFn: waitFn, killFn: killFn} + process := &execProcess{issue: issue, identity: identity, cmd: command, pid: pid, stdin: stdin, stdout: stdout, stderr: stderr, outputs: make(chan outputResult, 32), done: make(chan struct{}), started: started, waitFn: waitFn, killFn: killFn, releaseFn: releaseFn} manager.mu.Lock() manager.active[issue] = process manager.mu.Unlock() diff --git a/internal/client/supervisor/windows/native_other_test.go b/internal/client/supervisor/windows/native_other_test.go index 498bf8c..3a65980 100644 --- a/internal/client/supervisor/windows/native_other_test.go +++ b/internal/client/supervisor/windows/native_other_test.go @@ -8,6 +8,7 @@ import ( "errors" "io" "strings" + "sync/atomic" "testing" "time" @@ -16,6 +17,23 @@ import ( "github.com/rvbox/rvbox/internal/domain" ) +func TestExecProcessReleaseIsIdempotent_BH_LAUNCH_05(t *testing.T) { + var calls atomic.Int32 + process := &execProcess{releaseFn: func() error { + calls.Add(1) + return errors.New("release failed") + }} + if err := process.Release(context.Background()); err == nil || err.Error() != "release failed" { + t.Fatalf("first release error = %v", err) + } + if err := process.Release(context.Background()); err == nil || err.Error() != "release failed" { + t.Fatalf("repeated release error = %v", err) + } + if got := calls.Load(); got != 1 { + t.Fatalf("release calls = %d, want one", got) + } +} + func TestPortableSupervisorCapturesOutputAndSupportsStdin_HP_SUPERVISOR_03(t *testing.T) { t.Parallel() manager, err := NewSupervisor(NativeOptions{MaxOutputChunk: 8, WindowsTermGrace: 10 * time.Millisecond}) diff --git a/internal/client/supervisor/windows/native_windows.go b/internal/client/supervisor/windows/native_windows.go index f7e5c90..0a1fac4 100644 --- a/internal/client/supervisor/windows/native_windows.go +++ b/internal/client/supervisor/windows/native_windows.go @@ -188,7 +188,9 @@ func (manager *execSupervisor) Start(ctx context.Context, spec supervisor.StartS return fail(fmt.Errorf("create suspended command process: %w", err)) } // The child owns these handles after CreateProcessAsUser returns. Keep only - // the three parent ends and the process/job handles in the daemon. + // the three parent ends and the process/job handles in the daemon. The + // primary thread remains suspended until the executor has durably recorded + // launch authorization and calls Process.Release. _ = stdinRead.Close() _ = stdoutWrite.Close() _ = stderrWrite.Close() @@ -198,14 +200,19 @@ func (manager *execSupervisor) Start(ctx context.Context, spec supervisor.StartS _ = winapi.CloseHandle(processInfo.Thread) return fail(fmt.Errorf("assign command to Job: %w", err)) } - if _, err := winapi.ResumeThread(processInfo.Thread); err != nil { - _ = winapi.TerminateJobObject(job, 1) - _ = winapi.CloseHandle(processInfo.Process) - _ = winapi.CloseHandle(processInfo.Thread) - return fail(fmt.Errorf("release suspended command: %w", err)) - } pipesTransferred = true - _ = winapi.CloseHandle(processInfo.Thread) + var threadClosed sync.Once + closeThread := func() { + threadClosed.Do(func() { _ = winapi.CloseHandle(processInfo.Thread) }) + } + releaseFn := func() error { + if _, err := winapi.ResumeThread(processInfo.Thread); err != nil { + closeThread() + return fmt.Errorf("release suspended command: %w", err) + } + closeThread() + return nil + } started := manager.options.Now() handles := &nativeHandles{process: processInfo.Process, job: job, pid: processInfo.ProcessId} cleanupJob = false @@ -216,6 +223,7 @@ func (manager *execSupervisor) Start(ctx context.Context, spec supervisor.StartS if err := winapi.GetExitCodeProcess(processInfo.Process, &code); err != nil && waitErr == nil { waitErr = err } + closeThread() handles.close.Do(func() { _ = winapi.CloseHandle(processInfo.Process) _ = winapi.CloseHandle(job) @@ -223,9 +231,11 @@ func (manager *execSupervisor) Start(ctx context.Context, spec supervisor.StartS return int32(code), false, waitErr } killFn := func(code uint32) error { - return winapi.TerminateJobObject(job, code) + err := winapi.TerminateJobObject(job, code) + closeThread() + return err } - process := manager.registerProcess(spec.IssueUUID, identity, command, stdinWrite, stdoutRead, stderrRead, started, waitFn, killFn, cleanup) + process := manager.registerProcess(spec.IssueUUID, identity, command, stdinWrite, stdoutRead, stderrRead, started, waitFn, killFn, releaseFn, cleanup) process.snapshotFn = func() (supervisor.ResourceSnapshot, error) { return queryJobSnapshot(job, manager.options.Now()) } diff --git a/test/coverage.toml b/test/coverage.toml index 236de61..8eb1a12 100644 --- a/test/coverage.toml +++ b/test/coverage.toml @@ -500,6 +500,12 @@ layer = "unit" status = "implemented" tests = ["internal/agentproto/validate_test.go:TestWindowsEnvironmentCaseCollision_BH_LAUNCH_01"] +[[requirements]] +id = "BH-LAUNCH-05" +layer = "unit" +status = "implemented" +tests = ["internal/client/supervisor/windows/native_other_test.go:TestExecProcessReleaseIsIdempotent_BH_LAUNCH_05"] + [[requirements]] id = "BH-OUTFLOW-01" layer = "unit"