Honor cancellation before transcript delivery
This commit is contained in:
@@ -322,6 +322,8 @@ leave reports. Historical references continue to describe their own versions.
|
||||
|
||||
## R9 — P2: Honor Cancellation in Normalization-Only Processing
|
||||
|
||||
**Status:** Completed. Processing checks cancellation after normalization and before delivery. A controlled empty-pipeline test preserves the existing transcript, finalizes both YAML reports as failed, and retains the cancellation cause.
|
||||
|
||||
**Category**
|
||||
|
||||
Cancellation regression in Stages 8 and 12.
|
||||
|
||||
@@ -62,6 +62,9 @@ func runProcess(inv commandkit.Invocation, input processInput) commandkit.Outcom
|
||||
warn(debug.WriteNormalizedTranscript(normalized))
|
||||
warn(debug.WriteNormalizationSummary(summary))
|
||||
}
|
||||
if err := inv.Context.Err(); err != nil {
|
||||
return finishProcess(inv, input, loaded, prepared, reporting.Snapshot{}, nil, err)
|
||||
}
|
||||
result, runErr := executePipeline(inv.Context, loaded, prepared, normalized, warn)
|
||||
if runErr != nil {
|
||||
return finishProcess(inv, input, loaded, prepared, result.Snapshot, nil, runErr)
|
||||
|
||||
@@ -1,12 +1,15 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/redaction"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/outcomes"
|
||||
commandkit "go.maximumdirect.net/commandkit"
|
||||
)
|
||||
|
||||
@@ -21,15 +24,12 @@ func finishProcess(inv commandkit.Invocation, input processInput, loaded *proces
|
||||
completed := time.Now().UTC()
|
||||
snapshot.RunID, snapshot.StartedAt, snapshot.CompletedAt = prepared.RunID, prepared.Started, &completed
|
||||
snapshot.TranscriptPath, snapshot.OutputPath = input.Transcript, input.Output.Value
|
||||
if primary == nil {
|
||||
primary = inv.Context.Err()
|
||||
}
|
||||
snapshot.Status = reporting.StatusCompleted
|
||||
if primary != nil {
|
||||
snapshot.Status = reporting.StatusFailed
|
||||
snapshot = failedOutputSnapshot(snapshot, nil)
|
||||
if *snapshot.Summary.RunErrors == 0 {
|
||||
snapshot.Errors = append(snapshot.Errors, reporting.ErrorRecord{Scope: "run", Reason: "processing_failure", Message: "Processing failed"})
|
||||
snapshot = failedOutputSnapshot(snapshot, nil)
|
||||
}
|
||||
} else {
|
||||
snapshot.Status = reporting.StatusCompleted
|
||||
snapshot = failedProcessingSnapshot(snapshot, primary, inv.Context.Err())
|
||||
}
|
||||
reports, err := renderReportBundle(snapshot, secrets...)
|
||||
if err != nil {
|
||||
@@ -45,6 +45,11 @@ func finishProcess(inv commandkit.Invocation, input processInput, loaded *proces
|
||||
}
|
||||
return result
|
||||
}
|
||||
if canceled := inv.Context.Err(); primary == nil && canceled != nil {
|
||||
primary = canceled
|
||||
snapshot = failedProcessingSnapshot(snapshot, canceled, canceled)
|
||||
reports = failed(nil)
|
||||
}
|
||||
result := prepared.Prepared.Output.Finish(primary, transcript, reports, failed)
|
||||
if result != nil {
|
||||
return commandkit.Failure(redact(result)).WithCause(result)
|
||||
@@ -60,6 +65,22 @@ func finishProcess(inv commandkit.Invocation, input processInput, loaded *proces
|
||||
return commandkit.Success()
|
||||
}
|
||||
|
||||
func failedProcessingSnapshot(snapshot reporting.Snapshot, primary, contextErr error) reporting.Snapshot {
|
||||
for _, record := range snapshot.Errors {
|
||||
if record.Scope == "run" {
|
||||
return failedOutputSnapshot(snapshot, nil)
|
||||
}
|
||||
}
|
||||
record := reporting.ErrorRecord{Scope: "run", Reason: "processing_failure", Message: "Processing failed"}
|
||||
if errors.Is(primary, context.Canceled) || (contextErr != nil && errors.Is(primary, contextErr)) {
|
||||
record.Phase = "processing"
|
||||
record.Reason = string(outcomes.ReasonCanceled)
|
||||
record.Message = "Processing was canceled."
|
||||
}
|
||||
snapshot.Errors = append(snapshot.Errors, record)
|
||||
return failedOutputSnapshot(snapshot, nil)
|
||||
}
|
||||
|
||||
// failedOutputSnapshot derives the count from the same run records sent to each report.
|
||||
func failedOutputSnapshot(snapshot reporting.Snapshot, failures []diagnostics.FileFailure) reporting.Snapshot {
|
||||
snapshot.Status = reporting.StatusFailed
|
||||
|
||||
@@ -3,6 +3,7 @@ package cli
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -127,6 +128,70 @@ func TestDeliveryFailureSuppressesCompletionSummary(t *testing.T) {
|
||||
|
||||
}
|
||||
|
||||
func TestCanceledEmptyPipelineLeavesTranscriptAndFinalizesReports(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
for name, contents := range map[string]string{
|
||||
"config.yml": "version: 3\npipeline: {stages: []}\n",
|
||||
"transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`,
|
||||
"output.json": "previous transcript",
|
||||
} {
|
||||
if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
input := processInput{
|
||||
Config: commandkit.Supplied[string]{Value: filepath.Join(dir, "config.yml"), Set: true},
|
||||
Transcript: filepath.Join(dir, "transcript.json"),
|
||||
Output: commandkit.Supplied[string]{Value: filepath.Join(dir, "output.json"), Set: true},
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
cmd := commandkit.NewCommand(commandkit.CommandSpec[struct{}]{Name: "deliver", Run: func(inv commandkit.Invocation, _ struct{}) commandkit.Outcome {
|
||||
loaded, err := loadProcessInputs(inv.Context, input, inv.ReportingMode())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
prepared, err := prepareProcess(inv.Context, input, loaded, inv.Reporter().Output())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer prepared.Prepared.Output.Close()
|
||||
result, err := executePipeline(inv.Context, loaded, prepared, sourceToCanonicalTranscript(loaded.Source), func(error) {})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cancel()
|
||||
return finishProcess(inv, input, loaded, prepared, result.Snapshot, []byte("new transcript"), nil)
|
||||
}})
|
||||
app, err := commandkit.New(commandkit.Identity{Name: "test"}, cmd)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var stdout, stderr bytes.Buffer
|
||||
outcome := app.Execute(ctx, []string{"deliver", "--verbose"}, commandkit.Streams{Out: &stdout, Err: &stderr})
|
||||
if outcome.Status == 0 || !errors.Is(outcome.Cause, context.Canceled) || strings.Contains(stderr.String(), "Completed:") {
|
||||
t.Fatalf("outcome=%+v stderr=%q", outcome, stderr.String())
|
||||
}
|
||||
if data, err := os.ReadFile(input.Output.Value); err != nil || string(data) != "previous transcript" {
|
||||
t.Fatalf("transcript changed after cancellation: %q, %v", data, err)
|
||||
}
|
||||
var errorsDoc reporting.ErrorsDocument
|
||||
var recoveries reporting.RecoveriesDocument
|
||||
for name, target := range map[string]any{"errors.yml": &errorsDoc, "recoveries.yml": &recoveries} {
|
||||
data, err := os.ReadFile(filepath.Join(dir, name))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := yaml.Unmarshal(data, target); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if errorsDoc.Status != reporting.StatusFailed || recoveries.Status != reporting.StatusFailed ||
|
||||
len(errorsDoc.Errors) != 1 || errorsDoc.Errors[0].Reason != "canceled" {
|
||||
t.Fatalf("reports did not capture cancellation: %+v %+v", errorsDoc, recoveries)
|
||||
}
|
||||
}
|
||||
|
||||
type shortProcessWriter struct{}
|
||||
|
||||
func (shortProcessWriter) Write(data []byte) (int, error) { return len(data) / 2, nil }
|
||||
|
||||
Reference in New Issue
Block a user