eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 27 Wiki Activity

Separate subprocess execution lifecycle from policy

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-31 23:48:49 +00:00
parent 7284228bec
commit 04283e99d4
7 changed files with 120 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
internal/adapters/subprocess/diagnostics.go
View File

@@ -72,7 +72,7 @@ type logWriters struct {
Stdout io.Writer
Stderr io.Writer
limits chan *captureLimitError
limits chan error
mu sync.Mutex
limit *captureLimitError
stdout *diagnosticWriter
@@ -92,7 +92,7 @@ type diagnosticWriter struct {
}
func openLogWriters(stdoutPath, stderrPath, owner string, sensitiveValues []string) (*logWriters, error) {
logs := &logWriters{limits: make(chan *captureLimitError, 1)}
logs := &logWriters{limits: make(chan error, 1)}
cleanStdout := cleanLogPath(stdoutPath)
cleanStderr := cleanLogPath(stderrPath)
@@ -221,7 +221,7 @@ func (w *diagnosticWriter) reachLimit() error {
return limit
}
func (l *logWriters) Limits() <-chan *captureLimitError { return l.limits }
func (l *logWriters) Limits() <-chan error { return l.limits }
func (l *logWriters) Limit() *captureLimitError {
l.mu.Lock()

4
internal/adapters/subprocess/doc.go
View File

@@ -1,2 +1,4 @@
// Package subprocess provides reusable process execution and generated-config helpers.
// Package subprocess provides internal process execution and Narratio-specific
// subprocess policy. Its private execution mechanism owns process lifecycle;
// the Run boundary owns environments, diagnostics, redaction, and errors.
package subprocess

90
internal/adapters/subprocess/execution.go Normal file
View File

@@ -0,0 +1,90 @@
package subprocess
import (
"context"
"os/exec"
"time"
)
type executionReason uint8
const (
executionNotStarted executionReason = iota
executionNaturalExit
executionCanceled
executionDeadline
executionAborted
)
type executionOutcome struct {
reason executionReason
startedAt time.Time
completedAt time.Time
exitCode int
startErr error
waitErr error
contextErr error
abortErr error
cleanupErr error
}
func executeOwnedCommand(ctx context.Context, cmd *exec.Cmd, abort <-chan error) (outcome executionOutcome) {
outcome = executionOutcome{startedAt: time.Now().UTC(), exitCode: -1}
defer func() { outcome.completedAt = time.Now().UTC() }()
if outcome.contextErr = ctx.Err(); outcome.contextErr != nil {
outcome.reason = executionNotStarted
return outcome
}
tree, err := newOwnedProcessTree()
if err != nil {
outcome.reason, outcome.startErr = executionNotStarted, err
return outcome
}
defer func() { outcome.cleanupErr = joinErrors(outcome.cleanupErr, tree.Dispose()) }()
cmd.WaitDelay = forcefulTerminationWait
if outcome.startErr = tree.Start(cmd); outcome.startErr != nil {
outcome.reason = executionNotStarted
return outcome
}
waitCh := make(chan error, 1)
go func() { waitCh <- cmd.Wait() }()
select {
case outcome.waitErr = <-waitCh:
outcome.reason = executionNaturalExit
case <-ctx.Done():
outcome.contextErr = ctx.Err()
if outcome.contextErr == context.DeadlineExceeded {
outcome.reason = executionDeadline
} else {
outcome.reason = executionCanceled
}
outcome.cleanupErr = terminateAndReap(tree, waitCh)
case outcome.abortErr = <-abort:
outcome.reason = executionAborted
outcome.cleanupErr = terminateAndReap(tree, waitCh)
}
if cmd.ProcessState != nil {
outcome.exitCode = cmd.ProcessState.ExitCode()
}
return outcome
}
func terminateAndReap(tree ownedProcessTree, waitCh <-chan error) error {
cleanupErr := tree.TerminateGracefully()
timer := time.NewTimer(gracefulTerminationWait)
defer timer.Stop()
select {
case <-waitCh:
return joinErrors(cleanupErr, tree.TerminateForcefully())
case <-timer.C:
}
cleanupErr = joinErrors(cleanupErr, tree.TerminateForcefully())
forcefulTimer := time.NewTimer(forcefulTerminationWait)
defer forcefulTimer.Stop()
select {
case <-waitCh:
return cleanupErr
case <-forcefulTimer.C:
return joinErrors(cleanupErr, context.DeadlineExceeded)
}
}

71
internal/adapters/subprocess/run.go
View File

@@ -60,63 +60,23 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) {
}
defer logs.Close()
tree, err := newOwnedProcessTree()
if err != nil {
return RunResult{}, fmt.Errorf("prepare owned subprocess tree: %w", err)
}
cmd := exec.Command(req.Executable, req.Args...)
cmd.Dir = req.WorkingDir
cmd.Env = childEnv
cmd.Stdout = logs.Stdout
cmd.Stderr = logs.Stderr
// Streaming capture uses pipes. Bound their lifetime when a leader exits
// while a descendant still holds a stream descriptor.
cmd.WaitDelay = forcefulTerminationWait
started := time.Now().UTC()
result := RunResult{
ExitCode: -1,
StartedAt: started,
StdoutLogPath: req.StdoutLogPath,
StderrLogPath: req.StderrLogPath,
}
if err := runCtx.Err(); err != nil {
result.CompletedAt = time.Now().UTC()
result.Duration = result.CompletedAt.Sub(result.StartedAt)
return result, fmt.Errorf("command was not started: %w", err)
}
if err := tree.Start(cmd); err != nil {
result.CompletedAt = time.Now().UTC()
result.Duration = result.CompletedAt.Sub(result.StartedAt)
return result, fmt.Errorf("start command %q with args %v: %w", req.Executable, req.Args, err)
}
waitCh := make(chan error, 1)
go func() { waitCh <- cmd.Wait() }()
waitErr, ctxErr, captureLimit, cleanupErr := waitForOwnedCommand(runCtx, tree, waitCh, logs.Limits())
cleanupErr = joinErrors(cleanupErr, tree.Dispose())
cleanupErr = joinErrors(cleanupErr, logs.Flush())
outcome := executeOwnedCommand(runCtx, cmd, logs.Limits())
cleanupErr := joinErrors(outcome.cleanupErr, logs.Flush())
captureLimit, _ := outcome.abortErr.(*captureLimitError)
if captureLimit == nil {
captureLimit = logs.Limit()
}
result.CompletedAt = time.Now().UTC()
result := RunResult{ExitCode: outcome.exitCode, StartedAt: outcome.startedAt, CompletedAt: outcome.completedAt, StdoutLogPath: req.StdoutLogPath, StderrLogPath: req.StderrLogPath}
result.Duration = result.CompletedAt.Sub(result.StartedAt)
if cmd.ProcessState != nil {
result.ExitCode = cmd.ProcessState.ExitCode()
}
result.TimedOut = outcome.reason == executionDeadline
result.Canceled = outcome.reason == executionCanceled
if ctxErr == context.DeadlineExceeded {
result.TimedOut = true
}
if ctxErr == context.Canceled && !result.TimedOut {
result.Canceled = true
}
if waitErr == nil && ctxErr == nil && cleanupErr == nil {
if outcome.reason == executionNaturalExit && outcome.waitErr == nil && cleanupErr == nil {
return result, nil
}
@@ -124,25 +84,28 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) {
diagnostics := buildDiagnostics(req, result, stderrTail)
if captureLimit != nil {
if cause := joinErrors(waitErr, cleanupErr); cause != nil {
if cause := joinErrors(outcome.waitErr, cleanupErr); cause != nil {
return result, fmt.Errorf("%w (%s): %w", captureLimit, diagnostics, cause)
}
return result, fmt.Errorf("%w (%s)", captureLimit, diagnostics)
}
if result.TimedOut {
return result, fmt.Errorf("command timed out after %s (%s): %w", req.Timeout, diagnostics, joinErrors(ctxErr, waitErr, cleanupErr))
return result, fmt.Errorf("command timed out after %s (%s): %w", req.Timeout, diagnostics, joinErrors(outcome.contextErr, outcome.waitErr, cleanupErr))
}
if result.Canceled {
return result, fmt.Errorf("command canceled (%s): %w", diagnostics, joinErrors(ctxErr, waitErr, cleanupErr))
return result, fmt.Errorf("command canceled (%s): %w", diagnostics, joinErrors(outcome.contextErr, outcome.waitErr, cleanupErr))
}
if exitErr, ok := waitErr.(*exec.ExitError); ok {
return result, fmt.Errorf("command failed with exit code %d (%s): %w", exitErr.ExitCode(), diagnostics, joinErrors(waitErr, cleanupErr))
if exitErr, ok := outcome.waitErr.(*exec.ExitError); ok {
return result, fmt.Errorf("command failed with exit code %d (%s): %w", exitErr.ExitCode(), diagnostics, joinErrors(outcome.waitErr, cleanupErr))
}
if cleanupErr != nil {
return result, fmt.Errorf("command cleanup failed (%s): %w", diagnostics, joinErrors(waitErr, cleanupErr))
return result, fmt.Errorf("command cleanup failed (%s): %w", diagnostics, joinErrors(outcome.waitErr, cleanupErr))
}
return result, fmt.Errorf("command failed to run (%s): %w", diagnostics, joinErrors(waitErr, cleanupErr))
if outcome.startErr != nil {
return result, fmt.Errorf("start command %q with args %v: %w", req.Executable, req.Args, outcome.startErr)
}
return result, fmt.Errorf("command failed to run (%s): %w", diagnostics, joinErrors(outcome.waitErr, outcome.contextErr, outcome.abortErr, cleanupErr))
}
// WriteYAMLAtomic marshals value as YAML and atomically writes it to path.

6
internal/adapters/subprocess/run_test.go
View File

@@ -382,7 +382,7 @@ func TestStreamRedactorHandlesSplitAndOverlappingSecrets(t *testing.T) {
}
func TestDiagnosticWriterHonorsExactLimitAndCapPlusOne(t *testing.T) {
exactLogs := &logWriters{limits: make(chan *captureLimitError, 1)}
exactLogs := &logWriters{limits: make(chan error, 1)}
var exactOutput bytes.Buffer
exact := newDiagnosticWriter(exactLogs, "stdout", "test", &exactOutput, 5, nil)
if _, err := exact.Write([]byte("abcde")); err != nil {
@@ -398,7 +398,7 @@ func TestDiagnosticWriterHonorsExactLimitAndCapPlusOne(t *testing.T) {
t.Fatal("exact write recorded a capture limit")
}
cappedLogs := &logWriters{limits: make(chan *captureLimitError, 1)}
cappedLogs := &logWriters{limits: make(chan error, 1)}
var cappedOutput bytes.Buffer
capped := newDiagnosticWriter(cappedLogs, "stderr", "test", &cappedOutput, 5, nil)
if _, err := capped.Write([]byte("abcdef")); err == nil {
@@ -416,7 +416,7 @@ func TestDiagnosticWriterHonorsExactLimitAndCapPlusOne(t *testing.T) {
}
func TestDiagnosticWriterRetainsBoundedRedactedTail(t *testing.T) {
logs := &logWriters{limits: make(chan *captureLimitError, 1)}
logs := &logWriters{limits: make(chan error, 1)}
var output bytes.Buffer
secret := "credential-value"
writer := newDiagnosticWriter(logs, "stderr", "test", &output, 16*1024, []string{secret})