Create run workspaces only when needed
This commit is contained in:
@@ -45,6 +45,11 @@ file retains owner metadata after release or process death; its existence is
|
||||
not evidence that a lock is active. Command and restore flows wait for this
|
||||
lock only while their context remains active, and report a release failure.
|
||||
|
||||
The runner assigns every invocation a run-workspace identity but creates the
|
||||
physical directory only before an executing stage that requires it. Fully
|
||||
reused invocations and cleanup-only invocations therefore do not accumulate
|
||||
empty run directories.
|
||||
|
||||
## Cleanup Semantics
|
||||
|
||||
The canonical cleanup stage:
|
||||
|
||||
@@ -117,7 +117,10 @@ adapter and publish metadata.
|
||||
Resume and skip decisions are manifest-driven. Filesystem state may be
|
||||
inspected and validated, but file presence alone does not replace recorded
|
||||
stage state. Each invocation retains an in-memory command summary and a
|
||||
run-local working directory, but neither is a durable progress ledger.
|
||||
run-local workspace identity, but neither is a durable progress ledger. The
|
||||
physical run directory is created on demand only when an executing stage
|
||||
requires it; a fully reused invocation or a cleanup-only invocation does not
|
||||
leave an empty run directory.
|
||||
|
||||
Restore treats committed remote current state as its authority. It must plan
|
||||
deterministically, confine remote-to-local paths, protect local conflicts, and
|
||||
|
||||
@@ -207,6 +207,13 @@ func TestRunStageCleanupUsesPublishEvidenceAndFailsClosedWithoutIt(t *testing.T)
|
||||
t.Fatalf("cleanup target %q stat error = %v, want absent", target, statErr)
|
||||
}
|
||||
}
|
||||
cleaned := loadCleanupOrchestrationManifest(t, cfg)
|
||||
if cleaned.RunID == publishedRunID {
|
||||
t.Fatalf("cleanup invocation run id = %q, want a new invocation identity", cleaned.RunID)
|
||||
}
|
||||
if _, statErr := os.Lstat(cleaned.LocalWorkDir); !errors.Is(statErr, os.ErrNotExist) {
|
||||
t.Fatalf("cleanup invocation work directory stat error = %v, want absent", statErr)
|
||||
}
|
||||
|
||||
missingRoot := t.TempDir()
|
||||
missingPipeline, missingCampaign, missingSession := writeValidConfigFiles(t, missingRoot)
|
||||
|
||||
@@ -138,6 +138,7 @@ func TestPlannedStageModelMatchesStageExecution(t *testing.T) {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
at := time.Date(2026, time.May, 3, 12, 0, 0, 0, time.UTC)
|
||||
cfg := testConfig(t)
|
||||
workDir := filepath.Join(t.TempDir(), "run")
|
||||
cfg.Pipeline.Notarius = &config.NotariusConfig{Enabled: test.result == nil || test.result.Disposition != stage.StageDispositionSkipped}
|
||||
planned := manifest.New("2026-05-03", at)
|
||||
executed := manifest.New("2026-05-03", at)
|
||||
@@ -164,7 +165,7 @@ func TestPlannedStageModelMatchesStageExecution(t *testing.T) {
|
||||
ctx: context.Background(),
|
||||
cfg: cfg,
|
||||
opts: RunOptions{Force: test.force},
|
||||
identity: invocationIdentity{RunID: "20260503T120000Z-a1b2c3d4"},
|
||||
identity: invocationIdentity{RunID: "20260503T120000Z-a1b2c3d4", LocalWorkDir: workDir},
|
||||
env: &stage.Env{Config: cfg, Logger: slog.New(slog.NewTextHandler(io.Discard, nil))},
|
||||
manifestStore: &manifest.LocalStore{},
|
||||
manifestPath: filepath.Join(t.TempDir(), "manifest.json"),
|
||||
|
||||
@@ -157,15 +157,22 @@ func initializeLockedInvocation(runtime *invocationRuntime) error {
|
||||
if err := runtime.manifestStore.Save(runtime.ctx, runtime.manifestPath, m); err != nil {
|
||||
return fmt.Errorf("save manifest identity %q: %w", runtime.manifestPath, err)
|
||||
}
|
||||
if err := fileops.EnsureWorkspaceDirectory(runtime.identity.LocalWorkDir); err != nil {
|
||||
return runtime.persistTerminalFailure(fmt.Errorf("create run directory %q: %w", runtime.identity.LocalWorkDir, err))
|
||||
}
|
||||
if _, err := loadSecretsFromConfig(runtime.env.Config, runtime.env.Logger); err != nil {
|
||||
return runtime.persistTerminalFailure(fmt.Errorf("load secrets from files: %w", err))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (runtime *invocationRuntime) ensureRunWorkspaceFor(selected stage.Stage) error {
|
||||
if policy, ok := selected.(stage.RunWorkspacePolicy); ok && !policy.RequiresRunWorkspace() {
|
||||
return nil
|
||||
}
|
||||
if err := fileops.EnsureWorkspaceDirectory(runtime.identity.LocalWorkDir); err != nil {
|
||||
return fmt.Errorf("create run directory %q: %w", runtime.identity.LocalWorkDir, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (runtime *invocationRuntime) persistTerminalFailure(err error) error {
|
||||
return persistSessionTerminalFailure(
|
||||
runtime.ctx,
|
||||
|
||||
@@ -53,6 +53,9 @@ func executeSelectedStage(runtime *invocationRuntime, selected stage.Stage) (sta
|
||||
env.Logger.Info("skipping stage", "stage", selected.Name(), "reason", "already_succeeded", "force", runtime.opts.Force)
|
||||
return stageExecutionOutcome{skipped: true}, nil
|
||||
}
|
||||
if err := runtime.ensureRunWorkspaceFor(selected); err != nil {
|
||||
return stageExecutionOutcome{}, runtime.persistTerminalFailure(err)
|
||||
}
|
||||
|
||||
priorOutcome := capturePriorStageOutcome(m, selected.Name())
|
||||
attempted := stageExecutionOutcome{executed: true}
|
||||
|
||||
@@ -927,7 +927,7 @@ func TestExecuteStagesLoadsExistingManifest(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteStagesCreatesRunDirectoryPerInvocation(t *testing.T) {
|
||||
func TestExecuteStagesCreatesRunDirectoryForEachExecutedInvocation(t *testing.T) {
|
||||
cfg := testConfig(t)
|
||||
cfg.Pipeline.Spool.Root = t.TempDir()
|
||||
cfg.Pipeline.Storage.S3 = &config.StorageS3Config{
|
||||
@@ -972,6 +972,30 @@ func TestExecuteStagesCreatesRunDirectoryPerInvocation(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteStagesDoesNotCreateRunDirectoryWhenAllSelectedStagesAreReusable(t *testing.T) {
|
||||
cfg := testConfig(t)
|
||||
selected := []stage.Stage{resultStage{name: "notify"}}
|
||||
first, err := executeStages(context.Background(), cfg, selected, RunOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("first executeStages() error = %v", err)
|
||||
}
|
||||
second, err := executeStages(context.Background(), cfg, selected, RunOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("second executeStages() error = %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(second.Skipped, []string{"notify"}) {
|
||||
t.Fatalf("second skipped stages = %#v, want notify", second.Skipped)
|
||||
}
|
||||
firstWorkDir := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, first.RunID)
|
||||
if info, statErr := os.Stat(firstWorkDir); statErr != nil || !info.IsDir() {
|
||||
t.Fatalf("executed invocation work directory %q = %#v, %v; want directory", firstWorkDir, info, statErr)
|
||||
}
|
||||
skippedWorkDir := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, second.RunID)
|
||||
if _, statErr := os.Lstat(skippedWorkDir); !errors.Is(statErr, os.ErrNotExist) {
|
||||
t.Fatalf("skipped invocation work directory stat error = %v, want absent", statErr)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteStagesRejectsPersistedManifestIdentityMismatch(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
|
||||
@@ -28,6 +28,8 @@ type cleanupSemanticConfig struct {
|
||||
|
||||
func (cleanupStage) Name() string { return "cleanup" }
|
||||
|
||||
func (cleanupStage) RequiresRunWorkspace() bool { return false }
|
||||
|
||||
func (cleanupStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
|
||||
plan, err := buildCleanupPlan(env, m)
|
||||
if err != nil {
|
||||
|
||||
@@ -43,6 +43,12 @@ type Stage interface {
|
||||
Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error)
|
||||
}
|
||||
|
||||
// RunWorkspacePolicy lets a stage declare whether it requires the current
|
||||
// invocation's run directory. Stages require it by default.
|
||||
type RunWorkspacePolicy interface {
|
||||
RequiresRunWorkspace() bool
|
||||
}
|
||||
|
||||
const maxResumeReasonLength = 512
|
||||
|
||||
// ResumeValidation reports whether a previously succeeded stage can be reused.
|
||||
|
||||
Reference in New Issue
Block a user