Keep failed processing reports consistent
This commit is contained in:
@@ -45,6 +45,9 @@ validation, source publication, and asynchronous Linux delivery.
|
||||
|
||||
## R1 — P2: Keep Failure Records and Counts Consistent
|
||||
|
||||
**Status:** Completed. Failed report counts are derived from their records,
|
||||
and fatal attribution uses the returned processing error and known stage.
|
||||
|
||||
**Category**
|
||||
|
||||
Incorrect ordinary reports in Stages 7, 8, and 13.
|
||||
|
||||
@@ -96,10 +96,15 @@ func prepareProcess(ctx context.Context, input processInput, loaded *processInpu
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
snapshot.Status = reporting.StatusFailed
|
||||
failed, err := renderReportBundle(snapshot, secrets...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
failed := func(primary error, failures []diagnostics.FileFailure) diagnostics.ReportBundle {
|
||||
failedSnapshot := snapshot
|
||||
completed := time.Now().UTC()
|
||||
failedSnapshot.CompletedAt = &completed
|
||||
if primary != nil {
|
||||
failedSnapshot.Errors = []reporting.ErrorRecord{{Scope: "run", Phase: "initialization", Reason: "canceled", Message: "Processing was canceled."}}
|
||||
}
|
||||
bundle, _ := renderReportBundle(failedOutputSnapshot(failedSnapshot, failures), secrets...)
|
||||
return bundle
|
||||
}
|
||||
var debug *preflight.DebugInput
|
||||
var warnings []error
|
||||
|
||||
@@ -23,8 +23,10 @@ func finishProcess(inv commandkit.Invocation, input processInput, loaded *proces
|
||||
snapshot.TranscriptPath, snapshot.OutputPath = input.Transcript, input.Output.Value
|
||||
if primary != nil {
|
||||
snapshot.Status = reporting.StatusFailed
|
||||
if len(snapshot.Errors) == 0 {
|
||||
snapshot.Errors = []reporting.ErrorRecord{{Scope: "run", Reason: "processing_failure", Message: "Processing failed"}}
|
||||
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
|
||||
@@ -37,13 +39,7 @@ func finishProcess(inv commandkit.Invocation, input processInput, loaded *proces
|
||||
snapshot.Status = reporting.StatusFailed
|
||||
}
|
||||
failed := func(failures []diagnostics.FileFailure) diagnostics.ReportBundle {
|
||||
copySnapshot := snapshot
|
||||
copySnapshot.Status = reporting.StatusFailed
|
||||
copySnapshot.Errors = append([]reporting.ErrorRecord(nil), snapshot.Errors...)
|
||||
for _, failure := range failures {
|
||||
copySnapshot.Errors = append(copySnapshot.Errors, reporting.ErrorRecord{Scope: "run", Reason: "output_failure", Message: "Required output delivery failed", OutputPath: failure.Path})
|
||||
}
|
||||
result, bundleErr := renderReportBundle(copySnapshot, secrets...)
|
||||
result, bundleErr := renderReportBundle(failedOutputSnapshot(snapshot, failures), secrets...)
|
||||
if bundleErr != nil {
|
||||
return reports
|
||||
}
|
||||
@@ -63,3 +59,27 @@ func finishProcess(inv commandkit.Invocation, input processInput, loaded *proces
|
||||
}
|
||||
return commandkit.Success()
|
||||
}
|
||||
|
||||
// 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
|
||||
if len(failures) > 0 {
|
||||
snapshot.Errors = append([]reporting.ErrorRecord(nil), snapshot.Errors...)
|
||||
}
|
||||
for _, failure := range failures {
|
||||
record := reporting.ErrorRecord{Scope: "run", Reason: "output_failure", Message: "Required output delivery failed", OutputPath: failure.Path}
|
||||
if failure.Kind == "report initialization" {
|
||||
record.Phase = "initialization"
|
||||
record.Message = "Required report initialization failed"
|
||||
}
|
||||
snapshot.Errors = append(snapshot.Errors, record)
|
||||
}
|
||||
count := 0
|
||||
for _, record := range snapshot.Errors {
|
||||
if record.Scope == "run" {
|
||||
count++
|
||||
}
|
||||
}
|
||||
snapshot.Summary.RunErrors = &count
|
||||
return snapshot
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting"
|
||||
commandkit "go.maximumdirect.net/commandkit"
|
||||
"gopkg.in/yaml.v3"
|
||||
)
|
||||
|
||||
func TestPreparedDeliveryFollowsReportingMode(t *testing.T) {
|
||||
@@ -76,8 +77,9 @@ func TestPreparedDeliveryFollowsReportingMode(t *testing.T) {
|
||||
|
||||
func TestDeliveryFailureSuppressesCompletionSummary(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Chdir(dir)
|
||||
for name, contents := range map[string]string{
|
||||
"config.yml": "version: 3\npipeline:\n stages: []\nreporting:\n enabled: false\n",
|
||||
"config.yml": "version: 3\npipeline:\n stages: []\n",
|
||||
"transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`,
|
||||
} {
|
||||
if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0644); err != nil {
|
||||
@@ -105,6 +107,24 @@ func TestDeliveryFailureSuppressesCompletionSummary(t *testing.T) {
|
||||
if outcome.Status == 0 || strings.Contains(stderr.String(), "Completed:") {
|
||||
t.Fatalf("status=%d stderr=%q", outcome.Status, stderr.String())
|
||||
}
|
||||
var errs reporting.ErrorsDocument
|
||||
var recoveries reporting.RecoveriesDocument
|
||||
for path, target := range map[string]any{"errors.yml": &errs, "recoveries.yml": &recoveries} {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := yaml.Unmarshal(data, target); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if errs.Status != reporting.StatusFailed || recoveries.Status != reporting.StatusFailed ||
|
||||
errs.Summary.RunErrors == nil || *errs.Summary.RunErrors != 1 ||
|
||||
recoveries.Summary.RunErrors == nil || *recoveries.Summary.RunErrors != 1 ||
|
||||
len(errs.Errors) != 1 || errs.Errors[0].Reason != "output_failure" {
|
||||
t.Fatalf("failed output reports disagree: %+v %+v", errs, recoveries)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
type shortProcessWriter struct{}
|
||||
|
||||
@@ -2,6 +2,7 @@ package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics"
|
||||
@@ -13,6 +14,7 @@ import (
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/outcomes"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/processreport"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/runner"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/structuredoutput"
|
||||
)
|
||||
|
||||
func executePipeline(ctx context.Context, loaded *processInputs, prepared *preparedProcess, normalized *schema.Transcript, warn func(error)) (processResult, error) {
|
||||
@@ -47,22 +49,18 @@ func executePipeline(ctx context.Context, loaded *processInputs, prepared *prepa
|
||||
var failures []processreport.RunFailure
|
||||
if runErr != nil {
|
||||
failure := processreport.RunFailure{Phase: "processing", Reason: outcomes.ReasonInternalFailure}
|
||||
if ctx.Err() != nil {
|
||||
var cause *structuredoutput.Failure
|
||||
if errors.As(runErr, &cause) {
|
||||
failure.Reason = cause.Reason
|
||||
} else if errors.Is(runErr, context.Canceled) || (ctx.Err() != nil && errors.Is(runErr, ctx.Err())) {
|
||||
failure.Reason = outcomes.ReasonCanceled
|
||||
}
|
||||
if len(output.ModuleResults) > 0 {
|
||||
last := output.ModuleResults[len(output.ModuleResults)-1]
|
||||
index := last.StageIndex
|
||||
failure.StageIndex = &index
|
||||
origin := outcomes.Reference{StageIndex: &index, StageID: last.StageID, ModuleKey: last.ModuleKey, ModuleInstance: last.ModuleInstance}
|
||||
for _, operation := range last.Operations {
|
||||
if operation.FailureReason != "" {
|
||||
origin = operation.Reference
|
||||
failure.Reason = operation.FailureReason
|
||||
break
|
||||
}
|
||||
if last.Status == runner.ModuleStatusFailed {
|
||||
index := last.StageIndex
|
||||
failure.Origin = &outcomes.Reference{StageIndex: &index, StageID: last.StageID, ModuleKey: last.ModuleKey, ModuleInstance: last.ModuleInstance}
|
||||
}
|
||||
failure.Origin = &origin
|
||||
}
|
||||
failures = append(failures, failure)
|
||||
}
|
||||
|
||||
@@ -219,25 +219,34 @@ func (s *OutputSession) TranscriptPath() string {
|
||||
}
|
||||
|
||||
// Initialize publishes each initial report and attempts failed-status reports if any fail.
|
||||
func (s *OutputSession) Initialize(initial, failed ReportBundle) error {
|
||||
func (s *OutputSession) Initialize(initial ReportBundle, failed func([]FileFailure) ReportBundle) error {
|
||||
if s == nil || !s.enabled {
|
||||
return nil
|
||||
}
|
||||
var failures []error
|
||||
var failures []FileFailure
|
||||
for i, payload := range bundleParts(initial) {
|
||||
if err := s.replace(s.reports[i], payload); err != nil {
|
||||
failures = append(failures, FileFailure{Path: s.reports[i], Kind: "report initialization", Err: err})
|
||||
}
|
||||
}
|
||||
if len(failures) != 0 {
|
||||
for i, payload := range bundleParts(failed) {
|
||||
if len(failures) == 0 {
|
||||
return nil
|
||||
}
|
||||
if err := s.Close(); err != nil {
|
||||
failures = append(failures, FileFailure{Path: s.transcript, Kind: "output cleanup", Err: err})
|
||||
}
|
||||
if failed != nil {
|
||||
for i, payload := range bundleParts(failed(failures)) {
|
||||
if err := s.replace(s.reports[i], payload); err != nil {
|
||||
failures = append(failures, FileFailure{Path: s.reports[i], Kind: "failed report initialization", Err: err})
|
||||
}
|
||||
}
|
||||
failures = append(failures, s.Close())
|
||||
}
|
||||
return errors.Join(failures...)
|
||||
errs := make([]error, 0, len(failures))
|
||||
for _, failure := range failures {
|
||||
errs = append(errs, failure)
|
||||
}
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
// PrepareDebug creates optional initial diagnostics and returns any nonfatal problems.
|
||||
|
||||
@@ -121,7 +121,7 @@ func TestDisabledReportingIgnoresUnusableDirectory(t *testing.T) {
|
||||
if err != nil || len(session.ReportPaths()) != 0 {
|
||||
t.Fatalf("disabled reports probed: %+v %v", session, err)
|
||||
}
|
||||
if err := session.Initialize(ReportBundle{}, ReportBundle{}); err != nil {
|
||||
if err := session.Initialize(ReportBundle{}, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(dir, "missing")); !errors.Is(err, os.ErrNotExist) {
|
||||
@@ -147,7 +147,7 @@ func TestStdoutModeInitializesAndFinalizesReports(t *testing.T) {
|
||||
}
|
||||
initial := ReportBundle{Errors: []byte("in progress errors"), Recoveries: []byte("in progress recoveries"), Markdown: []byte("in progress report")}
|
||||
finished := ReportBundle{Errors: []byte("completed errors"), Recoveries: []byte("completed recoveries"), Markdown: []byte("completed report")}
|
||||
if err := session.Initialize(initial, ReportBundle{}); err != nil {
|
||||
if err := session.Initialize(initial, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if session.TranscriptPath() != "" {
|
||||
@@ -184,7 +184,12 @@ func TestInitializeFailureMarksPriorReportFailedBeforeModelWork(t *testing.T) {
|
||||
return actual(path, payload)
|
||||
}
|
||||
modelCalled := false
|
||||
if err := session.Initialize(ReportBundle{Errors: []byte("initial"), Recoveries: []byte("initial"), Markdown: []byte("initial")}, ReportBundle{Errors: []byte("failed"), Recoveries: []byte("failed"), Markdown: []byte("failed")}); err == nil {
|
||||
if err := session.Initialize(ReportBundle{Errors: []byte("initial"), Recoveries: []byte("initial"), Markdown: []byte("initial")}, func(failures []FileFailure) ReportBundle {
|
||||
if len(failures) != 1 || filepath.Base(failures[0].Path) != RecoveriesFile {
|
||||
t.Fatalf("initialization failures: %+v", failures)
|
||||
}
|
||||
return ReportBundle{Errors: []byte("failed"), Recoveries: []byte("failed"), Markdown: []byte("failed")}
|
||||
}); err == nil {
|
||||
modelCalled = true
|
||||
}
|
||||
if modelCalled {
|
||||
@@ -206,7 +211,7 @@ func TestFinalizationCollectsFailuresAndRetainsPublishedTranscript(t *testing.T)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := session.Initialize(ReportBundle{Errors: []byte("initial"), Recoveries: []byte("initial"), Markdown: []byte("initial")}, ReportBundle{}); err != nil {
|
||||
if err := session.Initialize(ReportBundle{Errors: []byte("initial"), Recoveries: []byte("initial"), Markdown: []byte("initial")}, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
actual := session.replace
|
||||
|
||||
@@ -19,7 +19,7 @@ type Input struct {
|
||||
Transcript *schema.SourceTranscript
|
||||
Output diagnostics.OutputSelection
|
||||
InitialReports diagnostics.ReportBundle
|
||||
FailedReports diagnostics.ReportBundle
|
||||
FailedReports func(error, []diagnostics.FileFailure) diagnostics.ReportBundle
|
||||
Debug *DebugInput
|
||||
}
|
||||
|
||||
@@ -65,18 +65,30 @@ func Prepare(ctx context.Context, input Input) (*Prepared, error) {
|
||||
_ = session.Close()
|
||||
return nil, err
|
||||
}
|
||||
if err := session.Initialize(input.InitialReports, input.FailedReports); err != nil {
|
||||
failedReports := func(primary error) func([]diagnostics.FileFailure) diagnostics.ReportBundle {
|
||||
return func(failures []diagnostics.FileFailure) diagnostics.ReportBundle {
|
||||
if input.FailedReports == nil {
|
||||
return diagnostics.ReportBundle{}
|
||||
}
|
||||
return input.FailedReports(primary, failures)
|
||||
}
|
||||
}
|
||||
if err := session.Initialize(input.InitialReports, failedReports(nil)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cancelOutput := func(err error) error {
|
||||
failed := failedReports(err)
|
||||
return session.Finish(err, nil, failed(nil), failed)
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, session.Finish(err, nil, input.FailedReports, func([]diagnostics.FileFailure) diagnostics.ReportBundle { return input.FailedReports })
|
||||
return nil, cancelOutput(err)
|
||||
}
|
||||
result := &Prepared{Output: session}
|
||||
if input.Debug != nil {
|
||||
result.Debug, result.Warnings = session.PrepareDebug(input.Debug.WorkDir, input.Debug.RunID, input.Debug.Metadata, input.Debug.EffectiveConfig, input.Debug.Secrets...)
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, session.Finish(err, nil, input.FailedReports, func([]diagnostics.FileFailure) diagnostics.ReportBundle { return input.FailedReports })
|
||||
return nil, cancelOutput(err)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
@@ -122,8 +122,10 @@ func Build(input BuildInput) reporting.Snapshot {
|
||||
if record.StageIndex == nil {
|
||||
record.StageIndex = failure.Origin.StageIndex
|
||||
}
|
||||
chunk := failure.Origin.SectionIndex
|
||||
record.ChunkIndex = &chunk
|
||||
if failure.Origin.Kind != "" {
|
||||
chunk := failure.Origin.SectionIndex
|
||||
record.ChunkIndex = &chunk
|
||||
}
|
||||
record.Operation = string(failure.Origin.Kind)
|
||||
record.Validator = failure.Origin.ValidatorName
|
||||
record.BatchIndex = failure.Origin.BatchIndex
|
||||
|
||||
Reference in New Issue
Block a user