From ff39b2fc4f9aedfd263b08a2d4dc90912109a3e0 Mon Sep 17 00:00:00 2001 From: Nishant Bangarwa Date: Thu, 23 Jul 2026 10:18:14 +0530 Subject: [PATCH] fix: preserve report delivery retry invariants --- runtime/reconcilers/report.go | 61 +++++--- runtime/reconcilers/report_delivery_test.go | 122 +++++++++++++++ runtime/reconcilers/report_test.go | 156 ++++++++++++++++++++ 3 files changed, 317 insertions(+), 22 deletions(-) create mode 100644 runtime/reconcilers/report_delivery_test.go create mode 100644 runtime/reconcilers/report_test.go diff --git a/runtime/reconcilers/report.go b/runtime/reconcilers/report.go index 315ee9cac6d0..2c9aaed8834a 100644 --- a/runtime/reconcilers/report.go +++ b/runtime/reconcilers/report.go @@ -466,16 +466,7 @@ func (r *ReportReconciler) executeSingle(ctx context.Context, self *runtimev1.Re // We're only going to retry on non-dirty cancellations. retry := false if reportErr != nil { - if errors.Is(reportErr, context.Canceled) || errors.Is(reportErr, context.DeadlineExceeded) { - if dirtyErr { - rep.State.CurrentExecution.ErrorMessage = "Report run was interrupted after some notifications were sent. The report will not automatically retry." - } else { - retry = true - rep.State.CurrentExecution.ErrorMessage = "Report run was interrupted. It will automatically retry." - } - } else { - rep.State.CurrentExecution.ErrorMessage = fmt.Sprintf("Report run failed: %v", reportErr.Error()) - } + retry, rep.State.CurrentExecution.ErrorMessage = classifyReportError(dirtyErr, reportErr) reportErr = fmt.Errorf("last report run failed with error: %v", reportErr.Error()) } @@ -495,6 +486,16 @@ func (r *ReportReconciler) executeSingle(ctx context.Context, self *runtimev1.Re return retry, reportErr } +func classifyReportError(dirty bool, err error) (bool, string) { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + if dirty { + return false, "Report run was interrupted after some notifications were sent. The report will not automatically retry." + } + return true, "Report run was interrupted. It will automatically retry." + } + return false, fmt.Sprintf("Report run failed: %v", err) +} + // sendReport composes and sends the actual report to the configured recipients. // It returns true if an error occurred after some or all notifications were sent. func (r *ReportReconciler) sendReport(ctx context.Context, self *runtimev1.Resource, rep *runtimev1.Report, t time.Time) (bool, []string, error) { @@ -591,33 +592,48 @@ func (r *ReportReconciler) sendReport(ctx context.Context, self *runtimev1.Resou } } - sent := false + sent, deliveryWarnings, err := r.sendReportNotifications(ctx, self, rep, t, notificationsContent) + allWarnings = append(allWarnings, deliveryWarnings...) + if err != nil { + return sent, allWarnings, err + } + + return sent, allWarnings, nil +} + +func (r *ReportReconciler) sendReportNotifications(ctx context.Context, self *runtimev1.Resource, rep *runtimev1.Report, t time.Time, notificationsContent map[string]*notificationData) (bool, []string, error) { + sentAny := false + var warnings []string for _, notifier := range rep.Spec.Notifiers { switch notifier.Connector { case "email": recipients := pbutil.ToSliceString(notifier.Properties.AsMap()["recipients"]) - sent, err = r.sendEmailNotification(ctx, self, rep, t, recipients, notificationsContent) + sent, notifierWarnings, err := r.sendEmailNotification(ctx, self, rep, t, recipients, notificationsContent) + sentAny = sentAny || sent + warnings = append(warnings, notifierWarnings...) if err != nil { - return sent, nil, err + return sentAny, warnings, err } default: - var err error - sent, err = r.sendNonEmailNotification(ctx, rep, t, notifier, notificationsContent) + sent, err := r.sendNonEmailNotification(ctx, rep, t, notifier, notificationsContent) + sentAny = sentAny || sent if err != nil { - return sent, nil, err + return sentAny, warnings, err } } } - return false, allWarnings, nil + return sentAny, warnings, nil } -func (r *ReportReconciler) sendEmailNotification(ctx context.Context, self *runtimev1.Resource, rep *runtimev1.Report, t time.Time, recipients []string, notificationsContent map[string]*notificationData) (bool, error) { +func (r *ReportReconciler) sendEmailNotification(ctx context.Context, self *runtimev1.Resource, rep *runtimev1.Report, t time.Time, recipients []string, notificationsContent map[string]*notificationData) (bool, []string, error) { sent := false + var warnings []string for _, recipient := range recipients { content, ok := notificationsContent[recipient] if !ok { r.C.Logger.Warn("Skipping recipient - failed to get notification content", zap.String("recipient", recipient), zap.String("report", self.Meta.Name.Name), observability.ZapCtx(ctx)) + warnings = append(warnings, fmt.Sprintf("Skipped recipient %q because notification content was unavailable", recipient)) continue } @@ -634,12 +650,12 @@ func (r *ReportReconciler) sendEmailNotification(ctx context.Context, self *runt DownloadFormat: content.downloadFormat, } if err := r.C.Runtime.Email.SendScheduledReport(opts); err != nil { - return true, fmt.Errorf("failed to send AI report email to %q: %w", recipient, err) + return sent, warnings, fmt.Errorf("failed to send AI report email to %q: %w", recipient, err) } sent = true } - return sent, nil + return sent, warnings, nil } // sendNonEmailNotification sends report via non-email notifiers (e.g., Slack). @@ -909,7 +925,7 @@ func calculateReportExecutionTimes(r *runtimev1.Report, watermark, previousWater // Calculate the execution times ts := []time.Time{end} - for i := 0; i < limit; i++ { + for len(ts) < limit { t := ts[len(ts)-1] t = d.Sub(t) if !t.After(previousWatermark) { @@ -926,7 +942,8 @@ func calculateReportExecutionTimes(r *runtimev1.Report, watermark, previousWater func latestReportWarnings(r *runtimev1.Report) []string { if r.State != nil && len(r.State.ExecutionHistory) > 0 { - return r.State.ExecutionHistory[len(r.State.ExecutionHistory)-1].Warnings + // Execution history is newest-first; popCurrentExecution inserts at index 0. + return r.State.ExecutionHistory[0].Warnings } return nil } diff --git a/runtime/reconcilers/report_delivery_test.go b/runtime/reconcilers/report_delivery_test.go new file mode 100644 index 000000000000..8523fe96639b --- /dev/null +++ b/runtime/reconcilers/report_delivery_test.go @@ -0,0 +1,122 @@ +package reconcilers + +import ( + "context" + "errors" + "testing" + "time" + + runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1" + "github.com/rilldata/rill/runtime" + "github.com/rilldata/rill/runtime/pkg/email" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "google.golang.org/protobuf/types/known/structpb" +) + +func TestReportEmailDeliveryTracksDirtyStateAcrossNotifiers(t *testing.T) { + // Automatic retry is safe only until the first successful external send. + // Separate email notifiers model the same state transition as email then Slack. + tests := []struct { + name string + failAt int + wantDirty bool + wantCalls []string + }{ + {name: "first delivery fails cleanly", failAt: 1, wantDirty: false, wantCalls: []string{"first@example.com"}}, + {name: "second delivery fails after side effect", failAt: 2, wantDirty: true, wantCalls: []string{"first@example.com", "second@example.com"}}, + {name: "all deliveries succeed", wantDirty: true, wantCalls: []string{"first@example.com", "second@example.com"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + sender := &failNthReportSender{failAt: tt.failAt, err: context.Canceled} + reconciler, self, report := newReportDeliveryFixture(t, sender, "first@example.com", "second@example.com") + content := map[string]*notificationData{ + "first@example.com": {openLink: "https://example.com/first"}, + "second@example.com": {openLink: "https://example.com/second"}, + } + + dirty, warnings, err := reconciler.sendReportNotifications(t.Context(), self, report, time.Date(2026, time.January, 2, 3, 4, 5, 0, time.UTC), content) + require.Equal(t, tt.wantDirty, dirty) + require.Empty(t, warnings) + require.Equal(t, tt.wantCalls, sender.recipients) + if tt.failAt == 0 { + require.NoError(t, err) + } else { + require.ErrorIs(t, err, context.Canceled) + } + }) + } +} + +func TestReportEmailDeliverySurfacesMissingRecipientContent(t *testing.T) { + // Recipient-mode metadata can omit a user. The run must record that omission + // instead of silently appearing fully successful while sending nothing. + sender := &failNthReportSender{} + reconciler, self, report := newReportDeliveryFixture(t, sender, "missing@example.com") + + dirty, warnings, err := reconciler.sendReportNotifications(t.Context(), self, report, time.Now(), map[string]*notificationData{}) + require.NoError(t, err) + require.False(t, dirty) + require.Empty(t, sender.recipients) + require.Equal(t, []string{`Skipped recipient "missing@example.com" because notification content was unavailable`}, warnings) +} + +func TestClassifyReportErrorRetryInvariant(t *testing.T) { + // Cancellation and deadline errors have identical retry semantics; ordinary + // failures are recorded but never treated as controller interruptions. + tests := []struct { + name string + dirty bool + err error + wantRetry bool + wantMessage string + }{ + {name: "clean cancellation retries", err: context.Canceled, wantRetry: true, wantMessage: "It will automatically retry"}, + {name: "dirty cancellation does not retry", dirty: true, err: context.Canceled, wantRetry: false, wantMessage: "will not automatically retry"}, + {name: "clean deadline retries", err: context.DeadlineExceeded, wantRetry: true, wantMessage: "It will automatically retry"}, + {name: "dirty deadline does not retry", dirty: true, err: context.DeadlineExceeded, wantRetry: false, wantMessage: "will not automatically retry"}, + {name: "ordinary failure does not retry", err: errors.New("export failed"), wantRetry: false, wantMessage: "Report run failed: export failed"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + retry, message := classifyReportError(tt.dirty, tt.err) + require.Equal(t, tt.wantRetry, retry) + require.Contains(t, message, tt.wantMessage) + }) + } +} + +type failNthReportSender struct { + failAt int + err error + recipients []string +} + +func (s *failNthReportSender) Send(toEmail, _ string, _, _ string) error { + s.recipients = append(s.recipients, toEmail) + if s.failAt > 0 && len(s.recipients) == s.failAt { + return s.err + } + return nil +} + +func newReportDeliveryFixture(t *testing.T, sender email.Sender, recipients ...string) (*ReportReconciler, *runtimev1.Resource, *runtimev1.Report) { + t.Helper() + controller := &runtime.Controller{ + Runtime: &runtime.Runtime{Email: email.New(sender)}, + Logger: zap.NewNop(), + } + reconciler := &ReportReconciler{C: controller} + self := &runtimev1.Resource{Meta: &runtimev1.ResourceMeta{Name: &runtimev1.ResourceName{Kind: runtime.ResourceKindReport, Name: "delivery-test"}}} + report := &runtimev1.Report{Spec: &runtimev1.ReportSpec{DisplayName: "Delivery test"}} + for _, recipient := range recipients { + props, err := structpb.NewStruct(map[string]any{"recipients": []any{recipient}}) + require.NoError(t, err) + report.Spec.Notifiers = append(report.Spec.Notifiers, &runtimev1.Notifier{ + Connector: "email", + Properties: props, + }) + } + return reconciler, self, report +} diff --git a/runtime/reconcilers/report_test.go b/runtime/reconcilers/report_test.go new file mode 100644 index 000000000000..a0dec012fcf5 --- /dev/null +++ b/runtime/reconcilers/report_test.go @@ -0,0 +1,156 @@ +package reconcilers + +import ( + "testing" + "time" + + runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1" + "github.com/stretchr/testify/require" +) + +func TestNextRefreshTimeBoundaries(t *testing.T) { + // Resolve disabled, ticker, cron, timezone, and invalid schedules at a fixed instant. + now := time.Date(2026, time.January, 2, 10, 30, 0, 0, time.UTC) + + // Schedules are persisted as the next instant strictly after now; this table + // covers disabled, ticker, cron, timezone, and precedence behavior. + tests := []struct { + name string + schedule *runtimev1.Schedule + want time.Time + wantErr string + }{ + {name: "nil schedule"}, + {name: "disabled", schedule: &runtimev1.Schedule{Disable: true}}, + {name: "ticker", schedule: &runtimev1.Schedule{TickerSeconds: 30}, want: now.Add(30 * time.Second)}, + {name: "cron", schedule: &runtimev1.Schedule{Cron: "0 * * * *"}, want: time.Date(2026, time.January, 2, 11, 0, 0, 0, time.UTC)}, + {name: "timezone cron", schedule: &runtimev1.Schedule{Cron: "0 6 * * *", TimeZone: "America/Los_Angeles"}, want: time.Date(2026, time.January, 2, 14, 0, 0, 0, time.UTC)}, + {name: "earliest of ticker and cron", schedule: &runtimev1.Schedule{TickerSeconds: 10, Cron: "0 * * * *"}, want: now.Add(10 * time.Second)}, + {name: "invalid cron", schedule: &runtimev1.Schedule{Cron: "not a cron"}, wantErr: "failed to parse cron schedule"}, + {name: "invalid timezone", schedule: &runtimev1.Schedule{Cron: "0 * * * *", TimeZone: "Mars/Olympus"}, wantErr: "failed to parse cron schedule"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := nextRefreshTime(now, tt.schedule) + if tt.wantErr != "" { + require.ErrorContains(t, err, tt.wantErr) + return + } + require.NoError(t, err) + require.Equal(t, tt.want, got) + }) + } +} + +func TestCalculateReportExecutionTimesWatermarkAndValidation(t *testing.T) { + // Watermark invariants prevent duplicate delivery, while malformed interval configuration must fail explicitly. + watermark := time.Date(2026, time.January, 10, 12, 0, 0, 0, time.UTC) + + t.Run("unchanged watermark is skipped", func(t *testing.T) { + // Re-running an inherited watermark would duplicate every downstream delivery. + _, err := calculateReportExecutionTimes(&runtimev1.Report{Spec: &runtimev1.ReportSpec{}}, watermark, watermark) + require.ErrorContains(t, err, "watermark is unchanged") + }) + + t.Run("no interval uses exact watermark", func(t *testing.T) { + // Reports without interval expansion must preserve the source watermark exactly. + got, err := calculateReportExecutionTimes(&runtimev1.Report{Spec: &runtimev1.ReportSpec{}}, watermark, time.Time{}) + require.NoError(t, err) + require.Equal(t, []time.Time{watermark}, got) + }) + + for _, tt := range []struct { + name string + interval string + timezone string + wantErr string + }{ + {name: "malformed duration", interval: "definitely-not-ISO-8601", wantErr: "failed to parse interval duration"}, + {name: "nonstandard duration", interval: "inf", wantErr: "is not a standard ISO 8601 duration"}, + {name: "invalid timezone", interval: "P1D", timezone: "Mars/Olympus", wantErr: "failed to load time zone"}, + } { + t.Run(tt.name, func(t *testing.T) { + // Parser validation is not a security boundary; reconciler inputs must still fail safely. + report := &runtimev1.Report{Spec: &runtimev1.ReportSpec{ + IntervalsIsoDuration: tt.interval, + RefreshSchedule: &runtimev1.Schedule{TimeZone: tt.timezone}, + }} + _, err := calculateReportExecutionTimes(report, watermark, time.Time{}) + require.ErrorContains(t, err, tt.wantErr) + }) + } +} + +func TestCalculateReportExecutionTimesFloorCeilAndLimit(t *testing.T) { + // Compare closed and open interval rounding and pin the exact execution-count limit. + watermark := time.Date(2026, time.January, 10, 12, 34, 0, 0, time.UTC) + previous := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) + + t.Run("closed intervals floor and respect the configured count", func(t *testing.T) { + // The limit counts executions, not loop iterations; returning limit+1 can + // unexpectedly fan out costly reports and notifications. + report := &runtimev1.Report{Spec: &runtimev1.ReportSpec{ + IntervalsIsoDuration: "P1D", + IntervalsLimit: 2, + }} + got, err := calculateReportExecutionTimes(report, watermark, previous) + require.NoError(t, err) + require.Equal(t, []time.Time{ + time.Date(2026, time.January, 9, 0, 0, 0, 0, time.UTC), + time.Date(2026, time.January, 10, 0, 0, 0, 0, time.UTC), + }, got) + }) + + t.Run("unclosed intervals ceil", func(t *testing.T) { + // Opting into an open interval rounds forward while still returning times chronologically. + report := &runtimev1.Report{Spec: &runtimev1.ReportSpec{ + IntervalsIsoDuration: "P1D", + IntervalsLimit: 2, + IntervalsCheckUnclosed: true, + }} + got, err := calculateReportExecutionTimes(report, watermark, previous) + require.NoError(t, err) + require.Equal(t, []time.Time{ + time.Date(2026, time.January, 10, 0, 0, 0, 0, time.UTC), + time.Date(2026, time.January, 11, 0, 0, 0, 0, time.UTC), + }, got) + }) + + t.Run("closed interval not beyond previous watermark is skipped", func(t *testing.T) { + // A partial interval at the same boundary must not be delivered twice. + report := &runtimev1.Report{Spec: &runtimev1.ReportSpec{IntervalsIsoDuration: "P1D"}} + _, err := calculateReportExecutionTimes(report, watermark, time.Date(2026, time.January, 10, 0, 0, 0, 0, time.UTC)) + require.ErrorContains(t, err, "watermark has not advanced by a full interval") + }) +} + +func TestCalculateReportExecutionTimesDSTBoundary(t *testing.T) { + // Local-midnight intervals span 23 hours across the New York spring-forward + // transition; fixed 24-hour arithmetic would select the wrong report dates. + report := &runtimev1.Report{Spec: &runtimev1.ReportSpec{ + IntervalsIsoDuration: "P1D", + IntervalsLimit: 2, + RefreshSchedule: &runtimev1.Schedule{TimeZone: "America/New_York"}, + }} + watermark := time.Date(2026, time.March, 9, 5, 30, 0, 0, time.UTC) + previous := time.Date(2026, time.March, 7, 5, 0, 0, 0, time.UTC) + got, err := calculateReportExecutionTimes(report, watermark, previous) + require.NoError(t, err) + require.Len(t, got, 2) + location, err := time.LoadLocation("America/New_York") + require.NoError(t, err) + require.True(t, time.Date(2026, time.March, 8, 0, 0, 0, 0, location).Equal(got[0])) + require.True(t, time.Date(2026, time.March, 9, 0, 0, 0, 0, location).Equal(got[1])) + require.Equal(t, 23*time.Hour, got[1].Sub(got[0])) +} + +func TestLatestReportWarningsUsesNewestHistoryEntry(t *testing.T) { + // Execution history is newest-first because popCurrentExecution inserts at + // index zero; surfacing the tail silently shows stale warnings to users. + report := &runtimev1.Report{State: &runtimev1.ReportState{ExecutionHistory: []*runtimev1.ReportExecution{ + {Warnings: []string{"newest warning"}}, + {Warnings: []string{"oldest warning"}}, + }}} + require.Equal(t, []string{"newest warning"}, latestReportWarnings(report)) + require.Nil(t, latestReportWarnings(&runtimev1.Report{})) +}