eric/weatherreporter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Preserve batch cancellation outcomes

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-13 03:02:59 +00:00
parent 4b748c2e53
commit 70cad789ea
11 changed files with 207 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
internal/app/app.go
View File

@@ -3,6 +3,7 @@ package app
import (
"context"
"errors"
"fmt"
"path/filepath"
"time"
@@ -100,6 +101,7 @@ type BatchResult struct {
Total int `json:"total"`
Succeeded int `json:"succeeded"`
Failed int `json:"failed"`
Canceled int `json:"canceled,omitempty"`
Notification *BatchNotificationResult `json:"notification,omitempty"`
Reports []BatchReportResult `json:"reports"`
}
@@ -143,12 +145,19 @@ type BatchReportResult struct {
type BatchError struct {
Result *BatchResult
Cause error
}
func (e BatchError) Error() string {
if e.Result == nil {
return "batch failed"
}
if errors.Is(e.Cause, context.DeadlineExceeded) {
return fmt.Sprintf("batch %s deadline exceeded", e.Result.Batch)
}
if errors.Is(e.Cause, context.Canceled) || e.Result.Canceled > 0 {
return fmt.Sprintf("batch %s canceled", e.Result.Batch)
}
failedReports := batchReportFailures(e.Result)
if batchNotificationFailed(e.Result) && failedReports == 0 {
if e.Result.Notification.Error != "" {
@@ -159,6 +168,10 @@ func (e BatchError) Error() string {
return fmt.Sprintf("batch %s failed: %d of %d reports failed", e.Result.Batch, failedReports, len(e.Result.Reports))
}
func (e BatchError) Unwrap() error {
return e.Cause
}
func batchNotificationFailed(result *BatchResult) bool {
return result != nil && result.Notification != nil && result.Notification.Status == "failed"
}
@@ -344,7 +357,12 @@ func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, erro
if req.Batch == BatchEvening || req.Batch == BatchMorning {
startedAt := now
result := &BatchResult{Batch: req.Batch, StartedAt: startedAt}
for _, planned := range plannedReports {
var cancellation error
for index, planned := range plannedReports {
if cancellation = batchCancellationCause(ctx, nil); cancellation != nil {
appendCanceledBatchReports(result, plannedReports[index:])
break
}
resolved := planned.Resolved
item := batchReportResult(planned)
reportResult, err := generatePromptReport(ctx, promptReportRequest{
@@ -364,14 +382,26 @@ func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, erro
copyBatchReportDetails(&item, reportResult)
}
if err != nil {
item.Status = "failed"
item.Error = err.Error()
result.Failed++
if cancellation = batchCancellationCause(ctx, err); cancellation != nil {
item.Status = "canceled"
result.Canceled++
} else {
item.Status = "failed"
item.Error = err.Error()
result.Failed++
}
} else {
item.Status = "succeeded"
result.Succeeded++
}
result.Reports = append(result.Reports, item)
if cancellation == nil {
cancellation = batchCancellationCause(ctx, nil)
}
if cancellation != nil {
appendCanceledBatchReports(result, plannedReports[index+1:])
break
}
}
result.Total = len(result.Reports)
batchNotification := notifyBatch(ctx, req.Config, req.Batch, batchRunID(startedAt, req.Batch), startedAt, result, plannedReports, req.Notifier)
@@ -379,11 +409,38 @@ func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, erro
result.Notification = batchNotification
}
result.FinishedAt = time.Now()
return result, nil
return result, cancellation
}
return nil, fmt.Errorf("run is not implemented")
}
func batchCancellationCause(ctx context.Context, err error) error {
if ctx != nil {
if contextErr := ctx.Err(); contextErr != nil {
return contextErr
}
}
if errors.Is(err, context.Canceled) {
return context.Canceled
}
if errors.Is(err, context.DeadlineExceeded) {
return context.DeadlineExceeded
}
return nil
}
func appendCanceledBatchReports(result *BatchResult, plannedReports []plannedBatchReport) {
if result == nil {
return
}
for _, planned := range plannedReports {
item := batchReportResult(planned)
item.Status = "canceled"
result.Reports = append(result.Reports, item)
result.Canceled++
}
}
func copyBatchReportDetails(item *BatchReportResult, result *ReportResult) {
item.LLMDebugPath = result.LLMDebugPath
item.OutputPath = result.OutputPath

61
internal/app/batch_generation_test.go
View File

@@ -21,7 +21,7 @@ func TestRunBatchDetailedKeepsSuccessfulOutputAndSkipsNotificationAfterPartialFa
Now: generationTime("2026-05-29T08:30:00-05:00"), WorkingDir: t.TempDir(), OutputDir: t.TempDir(),
Collector: &generationCollector{bundle: &bundle}, Executor: executor, Notifier: notifier,
})
if err != nil || result == nil || result.Total != 2 || result.Succeeded != 1 || result.Failed != 1 || result.Notification == nil || result.Notification.Status != "skipped" || notifier.batchCalls != 0 {
if err != nil || result == nil || result.Total != 2 || result.Succeeded != 1 || result.Failed != 1 || result.Canceled != 0 || result.Notification == nil || result.Notification.Status != "skipped" || notifier.batchCalls != 0 {
t.Fatalf("RunBatchDetailed() result/error/notifier = %#v/%v/%#v", result, err, notifier)
}
if result.Reports[0].Status != "succeeded" || result.Reports[0].OutputPath == "" || result.Reports[1].Status != "failed" || result.Reports[1].OutputPath != "" {
@@ -32,6 +32,65 @@ func TestRunBatchDetailedKeepsSuccessfulOutputAndSkipsNotificationAfterPartialFa
}
}
func TestRunBatchDetailedStopsAfterReportCancellation(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
bundle := generationBundle(t)
bundle.Hourly.Periods = bundle.Hourly.Periods[:1]
notifier := &generationNotifier{}
executor := &generationExecutor{cancelBeforeReturn: cancel}
result, err := RunBatchDetailed(ctx, BatchRequest{
Config: generationDistributorConfig(), Batch: BatchMorning,
Now: generationTime("2026-05-29T08:30:00-05:00"), WorkingDir: t.TempDir(), OutputDir: t.TempDir(),
Collector: &generationCollector{bundle: &bundle}, Executor: executor, Notifier: notifier,
})
if !errors.Is(err, context.Canceled) || result == nil || result.Total != 2 || result.Succeeded != 0 || result.Failed != 0 || result.Canceled != 2 || executor.executeCalls != 1 || notifier.batchCalls != 0 || result.Notification == nil || result.Notification.Status != "skipped" || result.Notification.Reason != "batch canceled" {
t.Fatalf("RunBatchDetailed() result/error/executor/notifier = %#v/%v/%#v/%#v", result, err, executor, notifier)
}
for _, item := range result.Reports {
if item.Status != "canceled" || item.OutputPath != "" {
t.Fatalf("canceled report = %#v", item)
}
}
}
func TestRunBatchDetailedRetainsPublishedReportBeforeCancellation(t *testing.T) {
for _, cause := range []error{context.Canceled, context.DeadlineExceeded} {
t.Run(cause.Error(), func(t *testing.T) {
bundle := generationBundle(t)
bundle.Hourly.Periods = bundle.Hourly.Periods[:1]
notifier := &generationNotifier{}
ctx := &publicationGateContext{Context: context.Background(), err: cause, afterChecks: 4}
result, err := RunBatchDetailed(ctx, BatchRequest{
Config: generationDistributorConfig(), Batch: BatchMorning,
Now: generationTime("2026-05-29T08:30:00-05:00"), WorkingDir: t.TempDir(), OutputDir: t.TempDir(),
Collector: &generationCollector{bundle: &bundle}, Executor: &generationExecutor{}, Notifier: notifier,
})
if !errors.Is(err, cause) || result == nil || result.Total != 2 || result.Succeeded != 1 || result.Failed != 0 || result.Canceled != 1 || len(result.Reports) != 2 || result.Reports[0].Status != "succeeded" || result.Reports[0].OutputPath == "" || result.Reports[1].Status != "canceled" || result.Reports[1].OutputPath != "" || notifier.batchCalls != 0 || result.Notification == nil || result.Notification.Status != "skipped" || result.Notification.Reason != "batch canceled" {
t.Fatalf("RunBatchDetailed() result/error/notifier = %#v/%v/%#v", result, err, notifier)
}
if _, statErr := os.Stat(result.Reports[0].OutputPath); statErr != nil {
t.Fatalf("published report %q: %v", result.Reports[0].OutputPath, statErr)
}
})
}
}
func TestRunBatchPreservesCancellationCause(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
bundle := generationBundle(t)
bundle.Hourly.Periods = bundle.Hourly.Periods[:1]
err := RunBatch(ctx, BatchRequest{
Config: generationDistributorConfig(), Batch: BatchMorning,
Now: generationTime("2026-05-29T08:30:00-05:00"), WorkingDir: t.TempDir(), OutputDir: t.TempDir(),
Collector: &generationCollector{bundle: &bundle}, Executor: &generationExecutor{cancelBeforeReturn: cancel}, Notifier: &generationNotifier{},
})
if !errors.Is(err, context.Canceled) {
t.Fatalf("RunBatch() error = %v", err)
}
}
func TestRunBatchDetailedNotifiesOnlyAfterAllOutputsExist(t *testing.T) {
bundle := generationBundle(t)
bundle.Hourly.Periods = bundle.Hourly.Periods[:1]

6
internal/app/batch_notification.go
View File

@@ -56,6 +56,12 @@ func notifyBatch(ctx context.Context, cfg config.Config, batch BatchKind, runID
if result == nil {
return failedBatchNotificationResult(batchNotificationRequest{}, fmt.Errorf("batch result is required"))
}
if result.Canceled > 0 {
return &BatchNotificationResult{
Status: "skipped",
Reason: "batch canceled",
}
}
if result.Failed > 0 {
return &BatchNotificationResult{
Status: "skipped",

11
internal/app/generation_test.go
View File

@@ -28,13 +28,18 @@ type generationCollector struct {
type publicationGateContext struct {
context.Context
err error
checks int
err error
checks int
afterChecks int
}
func (c *publicationGateContext) Err() error {
c.checks++
if c.checks >= 2 {
afterChecks := c.afterChecks
if afterChecks == 0 {
afterChecks = 2
}
if c.checks >= afterChecks {
return c.err
}
return nil