diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 2990429..d3accee 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -40,7 +40,7 @@ the agent policy; an intermediate compilation dependency is recorded accurately. | 2 | A02, A12: common Git mechanics | Complete — focused Git, candidate, identity, and public build/publication/delivery tests passed; A02 regression evidence follows in Stage 3. | | 3 | A02: source identity regression evidence | Complete — commit, tree, and blob replacement regressions and public publication, build, and delivery cases passed. | | 4 | A03: buffered command teardown | Complete — inherited-pipe, cancellation, healthy-command, native build, and publication recovery checks passed. | -| 5 | A03, A18: piped lifetime and queued cancellation | Pending | +| 5 | A03, A18: piped lifetime and queued cancellation | Complete — process and Git lifecycle tests passed under the race detector; public preparation cancellation passed. | | 6 | A04 | Pending | | 7 | A05 | Pending | | 8 | A06, A14 | Pending | diff --git a/internal/git/blob_reader.go b/internal/git/blob_reader.go index f455fe4..008814d 100644 --- a/internal/git/blob_reader.go +++ b/internal/git/blob_reader.go @@ -51,6 +51,9 @@ func (reader *BlobReader) Read(ctx context.Context, objectID string) ([]byte, er reader.mu.Lock() defer reader.mu.Unlock() + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("read blob %s: %w", objectID, err) + } if reader.terminal != nil { return nil, reader.terminal } diff --git a/internal/git/blob_reader_test.go b/internal/git/blob_reader_test.go index dd5ea7a..7f96c15 100644 --- a/internal/git/blob_reader_test.go +++ b/internal/git/blob_reader_test.go @@ -10,6 +10,7 @@ import ( "os/exec" "path/filepath" "strings" + "sync" "testing" "time" ) @@ -156,6 +157,46 @@ exec sleep 60`) } } +func TestBlobReaderCanceledQueuedReadKeepsSessionUsable(t *testing.T) { + fixture := newRemoteFixture(t) + object := gitText(t, fixture.work, "rev-parse", "HEAD:source.txt") + reader := NewBlobReader(t.Context(), fixture.work) + defer reader.Close() + if body, err := reader.Read(t.Context(), object); err != nil || string(body) != "candidate" { + t.Fatalf("first read = %q, %v", body, err) + } + reader.mu.Lock() + ctx, cancel := context.WithCancel(t.Context()) + observed := make(chan struct{}) + queued := &observedContext{Context: ctx, observed: observed} + result := make(chan error, 1) + go func() { + _, err := reader.Read(queued, object) + result <- err + }() + <-observed + cancel() + reader.mu.Unlock() + if err := <-result; !errors.Is(err, context.Canceled) { + t.Fatalf("queued read error = %v", err) + } + if body, err := reader.Read(t.Context(), object); err != nil || string(body) != "candidate" { + t.Fatalf("later read = %q, %v", body, err) + } +} + +type observedContext struct { + context.Context + once sync.Once + observed chan struct{} +} + +func (ctx *observedContext) Err() error { + err := ctx.Context.Err() + ctx.once.Do(func() { close(ctx.observed) }) + return err +} + func TestBlobReaderPreservesGitFailureAndStderr(t *testing.T) { installFakeGit(t, `printf 'fixture Git failure\n' >&2 exit 23`) diff --git a/internal/process/process.go b/internal/process/process.go index 7839d49..0a89ead 100644 --- a/internal/process/process.go +++ b/internal/process/process.go @@ -55,6 +55,8 @@ type PipedProcess struct { stdout io.ReadCloser parent context.Context cancel context.CancelFunc + stopIO func() bool + ioDone <-chan struct{} waitOnce sync.Once waitErr error } @@ -80,6 +82,9 @@ func (child *PipedProcess) Cancel() { func (child *PipedProcess) Wait() error { child.waitOnce.Do(func() { err := child.command.Wait() + if !child.stopIO() { + <-child.ioDone + } child.cancel() if err == nil { return @@ -121,12 +126,20 @@ func StartPiped(ctx context.Context, invocation Invocation) (*PipedProcess, erro cancel() return nil, fmt.Errorf("start %q: %w", invocation.Args[0], err) } + done := make(chan struct{}) + stopIO := context.AfterFunc(childContext, func() { + defer close(done) + _ = stdin.Close() + _ = stdout.Close() + }) return &PipedProcess{ command: command, stdin: stdin, stdout: stdout, parent: ctx, cancel: cancel, + stopIO: stopIO, + ioDone: done, }, nil } diff --git a/internal/process/process_test.go b/internal/process/process_test.go index 6d54cf1..7a35167 100644 --- a/internal/process/process_test.go +++ b/internal/process/process_test.go @@ -260,6 +260,52 @@ func TestStartPipedCancellationAfterReadinessReapsChild(t *testing.T) { } } +func TestStartPipedCancellationInterruptsInheritedOutputDrain(t *testing.T) { + ready := filepath.Join(t.TempDir(), "descendant") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + child, err := StartPiped(ctx, Invocation{ + Args: []string{helperExecutable(t), "-test.run=TestProcessHelper", "--", "unused", "exit-descendant"}, + Env: map[string]string{"RELEASEKIT_PROCESS_HELPER": "1", "RELEASEKIT_DESCENDANT_READY": ready}, + }) + if err != nil { + t.Fatal(err) + } + _ = child.Stdin().Close() + drained := make(chan error, 1) + go func() { + _, err := io.Copy(io.Discard, child.Stdout()) + drained <- err + }() + deadline := time.After(10 * time.Second) + var pid int + for pid == 0 { + if data, err := os.ReadFile(ready); err == nil { + pid, _ = strconv.Atoi(string(data)) + } + select { + case <-deadline: + t.Fatal("descendant did not become ready") + default: + time.Sleep(10 * time.Millisecond) + } + } + t.Cleanup(func() { + if descendant, err := os.FindProcess(pid); err == nil { + _ = descendant.Kill() + } + }) + cancel() + select { + case <-drained: + case <-time.After(10 * time.Second): + t.Fatal("inherited stdout did not close after cancellation") + } + if err := child.Wait(); err != nil && !errors.Is(err, context.Canceled) { + t.Fatalf("wait after cancellation: %v", err) + } +} + func TestStartPipedCancelDoesNotCancelParentContext(t *testing.T) { ctx := context.Background() child, err := StartPiped(ctx, Invocation{ diff --git a/preparation_test.go b/preparation_test.go index 00bdbb3..25a69cd 100644 --- a/preparation_test.go +++ b/preparation_test.go @@ -6,12 +6,58 @@ import ( "os" "os/exec" "path/filepath" + "strconv" "strings" "testing" + "time" releasekit "go.maximumdirect.net/releasekit" ) +func TestPrepareBuildDependenciesCancellationReleasesInheritedOutput(t *testing.T) { + directory := t.TempDir() + ready := filepath.Join(t.TempDir(), "descendant") + gitPath := filepath.Join(t.TempDir(), "git") + if err := os.WriteFile(gitPath, []byte("#!/bin/sh\nsleep 60 &\nprintf '%s' \"$!\" > \"$RELEASEKIT_PREPARE_DESCENDANT\"\nwait\n"), 0o700); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", filepath.Dir(gitPath)+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("RELEASEKIT_PREPARE_DESCENDANT", ready) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + result := make(chan error, 1) + go func() { + result <- releasekit.PrepareBuildDependencies(ctx, releasekit.DependencyPreparationRequest{RepositoryDir: directory}) + }() + deadline := time.After(10 * time.Second) + var pid int + for pid == 0 { + if data, err := os.ReadFile(ready); err == nil { + pid, _ = strconv.Atoi(string(data)) + } + select { + case <-deadline: + t.Fatal("preparation descendant did not start") + default: + time.Sleep(10 * time.Millisecond) + } + } + t.Cleanup(func() { + if descendant, err := os.FindProcess(pid); err == nil { + _ = descendant.Kill() + } + }) + cancel() + select { + case err := <-result: + if !errors.Is(err, context.Canceled) { + t.Fatalf("preparation error = %v, want cancellation", err) + } + case <-time.After(10 * time.Second): + t.Fatal("preparation did not return after cancellation") + } +} + func TestPrepareTaggedCheckoutFetchesExpectedCommitAndDetaches(t *testing.T) { repository := newTaggedDeliveryConsumer(t) tagCommit := publishPreparationTag(t, repository)