diff --git a/app/common/billing.go b/app/common/billing.go index 2b74d51d85..ab3b6524cf 100644 --- a/app/common/billing.go +++ b/app/common/billing.go @@ -250,6 +250,7 @@ func NewBillingRegistry( if creditsConfig.Enabled { chargesRegistry, err = newChargesRegistry( + metricMeter, db, logger, locker, diff --git a/app/common/charges.go b/app/common/charges.go index 5a05ecbdcd..e934fa1ae1 100644 --- a/app/common/charges.go +++ b/app/common/charges.go @@ -4,6 +4,8 @@ import ( "fmt" "log/slog" + "go.opentelemetry.io/otel/metric" + "github.com/openmeterio/openmeter/app/config" "github.com/openmeterio/openmeter/openmeter/billing" "github.com/openmeterio/openmeter/openmeter/billing/charges" @@ -64,6 +66,7 @@ func NewChargesMetaAdapter( } func NewChargesCollectorService( + metricMeter metric.Meter, logger *slog.Logger, db *entdb.Client, ledgerService ledger.Ledger, @@ -74,6 +77,7 @@ func NewChargesCollectorService( breakageService ledgerbreakage.Service, ) (ledgercollector.Service, error) { collectorService, err := ledgercollector.NewService(ledgercollector.Config{ + Meter: metricMeter, Logger: logger, Advance: advanceService, Ledger: ledgerService, @@ -435,6 +439,7 @@ func NewRecognizerService( // newChargesRegistry constructs the full charges stack. // Private: must be initialized via NewBillingRegistry. func newChargesRegistry( + metricMeter metric.Meter, db *entdb.Client, logger *slog.Logger, locker *lockr.Locker, @@ -479,6 +484,7 @@ func newChargesRegistry( transactionManager := enttx.NewCreator(db) collectorService, err := ledgercollector.NewService(ledgercollector.Config{ + Meter: metricMeter, Logger: logger, Advance: advanceService, Ledger: ledgerService, diff --git a/openmeter/billing/README.md b/openmeter/billing/README.md index 8fe47b68d5..fda4f810bc 100644 --- a/openmeter/billing/README.md +++ b/openmeter/billing/README.md @@ -200,6 +200,11 @@ earlier attempt remain available to idempotent retry handling. ## Validation issues alongside successful results +`models.Error` carries codes and diagnostic attributes but remains a system error. +`models.ValidationIssue` explicitly opts into validation handling. Their shared +metadata does not make them interchangeable; only an explicit +`WrapAsValidationIssue` conversion changes a system error into a billing issue. + Typed validation issues can accompany usable results through an `error` return. Classify that error before discarding the result. Its meaning depends on the boundary: diff --git a/openmeter/billing/charges/service/base_test.go b/openmeter/billing/charges/service/base_test.go index 120106f415..cdb3b396ab 100644 --- a/openmeter/billing/charges/service/base_test.go +++ b/openmeter/billing/charges/service/base_test.go @@ -9,6 +9,7 @@ import ( "github.com/alpacahq/alpacadecimal" "github.com/invopop/gobl/currency" "github.com/samber/lo" + metricnoop "go.opentelemetry.io/otel/metric/noop" "github.com/openmeterio/openmeter/openmeter/billing" "github.com/openmeterio/openmeter/openmeter/billing/charges" @@ -187,6 +188,7 @@ func (s *BaseSuite) SetupSuite() { advanceService := advancetestutils.NewService(s.T(), ledgerDeps, breakageService) collectorService, err := ledgercollector.NewService(ledgercollector.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: omtestutils.NewDiscardLogger(s.T()), Advance: advanceService, Ledger: ledgerDeps.HistoricalLedger, diff --git a/openmeter/billing/validationissue.go b/openmeter/billing/validationissue.go index c851912504..b3a1a4429f 100644 --- a/openmeter/billing/validationissue.go +++ b/openmeter/billing/validationissue.go @@ -529,14 +529,33 @@ func toValidationIssue(err error, fieldPrefix string, component ComponentName, m Attributes: mergedAttributes, }, }, nil + case models.Error: + // Generic errors remain system errors unless explicitly converted with WrapAsValidationIssue. + if ordinaryErrorSeverity == "" { + return nil, err + } + + // Conversion is explicit at this boundary; keep the structured context. + severity := models.ErrorSeverityCritical + if ordinaryErrorSeverity == ValidationIssueSeverityWarning { + severity = models.ErrorSeverityWarning + } + + return modelErrorAsBillingValidationIssue(errT.WithSeverity(severity), fieldPrefix, component, messagePrefix, attributes) + case *models.Error: + if errT == nil { + return nil, nil + } + + return toValidationIssue(*errT, fieldPrefix, component, messagePrefix, attributes, ordinaryErrorSeverity) case models.ValidationIssue: - return modelValidationIssueAsBillingValidationIssue(errT, fieldPrefix, component, messagePrefix, attributes) + return modelErrorAsBillingValidationIssue(models.Error(errT), fieldPrefix, component, messagePrefix, attributes) case *models.ValidationIssue: if errT == nil { return nil, nil } - return modelValidationIssueAsBillingValidationIssue(*errT, fieldPrefix, component, messagePrefix, attributes) + return modelErrorAsBillingValidationIssue(models.Error(*errT), fieldPrefix, component, messagePrefix, attributes) } switch errT := err.(type) { @@ -578,7 +597,7 @@ func toValidationIssue(err error, fieldPrefix string, component ComponentName, m } } -func modelValidationIssueAsBillingValidationIssue(issue models.ValidationIssue, fieldPrefix string, component ComponentName, messagePrefix string, attributes models.Annotations) ([]ValidationIssue, error) { +func modelErrorAsBillingValidationIssue(issue models.Error, fieldPrefix string, component ComponentName, messagePrefix string, attributes models.Annotations) ([]ValidationIssue, error) { issueComponent := component if issueComponent == "" { issueComponent = ComponentName(issue.Component()) diff --git a/openmeter/billing/validationissue_test.go b/openmeter/billing/validationissue_test.go index 86b10cf19e..dcbd034cfe 100644 --- a/openmeter/billing/validationissue_test.go +++ b/openmeter/billing/validationissue_test.go @@ -872,3 +872,43 @@ func TestValidationWithMessage(t *testing.T) { require.NoError(t, ValidationWithMessage(nil, "unused", models.Attributes{"unused": true})) }) } + +func TestStructuredSystemErrorClassification(t *testing.T) { + validation := models.NewValidationIssue("invoice_invalid", "invoice cannot proceed") + system := models.NewError("ledger_failed", "accounting failed", models.WithCause(validation)). + WithAttr("reason", "missing_source") + + for _, tc := range []struct { + name string + err error + }{ + {name: "value", err: system}, + {name: "pointer", err: &system}, + {name: "formatted wrapper", err: fmt.Errorf("collect: %w", system)}, + {name: "component wrapper", err: ValidationWithComponent("ledger", system)}, + {name: "field wrapper", err: ValidationWithFieldPrefix("lines", system)}, + {name: "attribute wrapper", err: ValidationWithAttributes(models.Annotations{"operation": "collect"}, system)}, + {name: "joined with warning", err: errors.Join(NewValidationWarning("warning", "warning"), system)}, + } { + t.Run(tc.name, func(t *testing.T) { + issues, err := ToValidationIssues(tc.err) + require.ErrorIs(t, err, system) + require.Empty(t, issues) + require.False(t, IsValidationIssueOnly(tc.err)) + }) + } +} + +func TestExplicitStructuredErrorConversionPreservesContext(t *testing.T) { + system := models.NewError("provider_failed", "provider rejected invoice", models.WithComponent("provider")). + WithAttr("reason", "missing_address") + + issues, err := ToValidationIssues(WrapAsValidationIssue(fmt.Errorf("sync: %w", system), WithWarningSeverity())) + require.NoError(t, err) + require.Len(t, issues, 1) + require.Equal(t, "provider_failed", issues[0].Code) + require.Equal(t, system.Message(), issues[0].Message) + require.Equal(t, ComponentName("provider"), issues[0].Component) + require.Equal(t, ValidationIssueSeverityWarning, issues[0].Severity) + require.Equal(t, "missing_address", issues[0].Attributes["reason"]) +} diff --git a/openmeter/billing/worker/subscriptionsync/service/sync_credittheninvoice_test.go b/openmeter/billing/worker/subscriptionsync/service/sync_credittheninvoice_test.go index c4a9179832..d0e8c5a056 100644 --- a/openmeter/billing/worker/subscriptionsync/service/sync_credittheninvoice_test.go +++ b/openmeter/billing/worker/subscriptionsync/service/sync_credittheninvoice_test.go @@ -13,6 +13,7 @@ import ( "github.com/samber/mo" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" + metricnoop "go.opentelemetry.io/otel/metric/noop" "github.com/openmeterio/openmeter/openmeter/billing" "github.com/openmeterio/openmeter/openmeter/billing/charges" @@ -86,6 +87,7 @@ func (s *CreditThenInvoiceTestSuite) SetupSuite() { advanceService := advancetestutils.NewService(s.T(), ledgerDeps, breakageService) collectorService, err := ledgercollector.NewService(ledgercollector.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: logger, Advance: advanceService, Ledger: ledgerDeps.HistoricalLedger, diff --git a/openmeter/ledger/README.md b/openmeter/ledger/README.md index d01d133018..8bf71dcb1d 100644 --- a/openmeter/ledger/README.md +++ b/openmeter/ledger/README.md @@ -87,6 +87,17 @@ persist the returned group reference with its own lifecycle state. Ledger annotations and entry identity preserve accounting meaning and provenance; they are not operation idempotency keys. +## Error diagnostics + +Accounting guards use fixed diagnostic reasons, such as +`ledger_breakage_release_exceeds_open_amount`. Add them with +`models.NewError` and its `reason` attribute. These remain system errors; +use `models.NewValidationIssue` only for domain validation. +Existing validation issues keep their codes and `reason` attributes. +The shared error handler logs both as `error_codes` and `error_reasons`, including +through wrapping and joined errors. Keep IDs and amounts in the message, not the +reason. Direct logger calls do not extract this metadata. + ## Collection provenance Entry provenance carries three distinct identities: diff --git a/openmeter/ledger/advance/service/attribution.go b/openmeter/ledger/advance/service/attribution.go index a02f29e69d..71ff482400 100644 --- a/openmeter/ledger/advance/service/attribution.go +++ b/openmeter/ledger/advance/service/attribution.go @@ -12,6 +12,7 @@ import ( "github.com/openmeterio/openmeter/openmeter/ledger" "github.com/openmeterio/openmeter/pkg/cmpx" "github.com/openmeterio/openmeter/pkg/currencyx" + "github.com/openmeterio/openmeter/pkg/models" ) // advanceAttribution keeps the amounts and original routes needed to attribute @@ -113,7 +114,7 @@ func allocateAccruedBackedAdvanceAttributions( } }) if allocation.Amount.Sub(consumed).IsPositive() { - return nil, fmt.Errorf("advance attribution allocation %s exceeds remaining receivable for spend charge", allocation.Amount.String()) + return nil, models.NewError(ledger.ErrCodeAdvanceFailed, fmt.Sprintf("advance attribution allocation %s exceeds remaining receivable for spend charge", allocation.Amount.String()), models.WithAttribute("reason", "ledger_advance_attribution_exceeds_receivable")) } attributions = append(attributions, allocated...) diff --git a/openmeter/ledger/advance/service/legacy_correction.go b/openmeter/ledger/advance/service/legacy_correction.go index cfc43e137c..add1655e9e 100644 --- a/openmeter/ledger/advance/service/legacy_correction.go +++ b/openmeter/ledger/advance/service/legacy_correction.go @@ -61,7 +61,7 @@ func (s *service) PlanLegacyCorrection(ctx context.Context, input advance.Legacy } if attribution == nil { - return out, fmt.Errorf("backing group %s has no receivable attribution matching the collected source", group.ID().ID) + return out, models.NewError(ledger.ErrCodeAdvanceFailed, fmt.Sprintf("backing group %s has no receivable attribution matching the collected source", group.ID().ID), models.WithAttribute("reason", "ledger_advance_receivable_attribution_missing")) } out.LegacyCorrections = append(out.LegacyCorrections, transactions.CorrectionInput{ @@ -272,5 +272,5 @@ func (s *service) backfilledCreditReissueRoute(group ledger.TransactionGroup) (b }, nil } - return backfilledCreditReissueRouteResult{}, fmt.Errorf("backing transaction group %s does not contain a known cost basis route", group.ID().ID) + return backfilledCreditReissueRouteResult{}, models.NewError(ledger.ErrCodeAdvanceFailed, fmt.Sprintf("backing transaction group %s does not contain a known cost basis route", group.ID().ID), models.WithAttribute("reason", "ledger_advance_cost_basis_route_missing")) } diff --git a/openmeter/ledger/breakage/breakage_impacts.go b/openmeter/ledger/breakage/breakage_impacts.go index 724233305d..76b70a744b 100644 --- a/openmeter/ledger/breakage/breakage_impacts.go +++ b/openmeter/ledger/breakage/breakage_impacts.go @@ -94,7 +94,7 @@ func (s *service) ListExpiredBreakageImpacts(ctx context.Context, input ListExpi reference := subAccount.Route().Currency.Clone() if reference.GetCode() != record.Currency { - return ListExpiredBreakageImpactsResult{}, fmt.Errorf("breakage FBO sub-account %s currency %s does not match record currency %s", record.FBOSubAccountID, reference.GetCode(), record.Currency) + return ListExpiredBreakageImpactsResult{}, models.NewError(ledger.ErrCodeBreakageFailed, fmt.Sprintf("breakage FBO sub-account %s currency %s does not match record currency %s", record.FBOSubAccountID, reference.GetCode(), record.Currency), models.WithAttribute("reason", "ledger_breakage_currency_mismatch")) } currencyReferences[record.FBOSubAccountID] = reference @@ -150,11 +150,11 @@ func (s *service) ListExpiredBreakageImpacts(ctx context.Context, input ListExpi } if group.amount.IsNegative() { - return ListExpiredBreakageImpactsResult{}, fmt.Errorf("expired breakage amount is negative for %s %s", group.expiresAt, group.currency) + return ListExpiredBreakageImpactsResult{}, models.NewError(ledger.ErrCodeBreakageFailed, fmt.Sprintf("expired breakage amount is negative for %s %s", group.expiresAt, group.currency), models.WithAttribute("reason", "ledger_breakage_expired_amount_negative")) } if group.cursorID.ID == "" { - return ListExpiredBreakageImpactsResult{}, fmt.Errorf("expired breakage impact has no plan record for %s %s", group.expiresAt, group.currency) + return ListExpiredBreakageImpactsResult{}, models.NewError(ledger.ErrCodeBreakageFailed, fmt.Sprintf("expired breakage impact has no plan record for %s %s", group.expiresAt, group.currency), models.WithAttribute("reason", "ledger_breakage_plan_missing")) } annotations := models.Annotations{ diff --git a/openmeter/ledger/breakage/service.go b/openmeter/ledger/breakage/service.go index 64527be4e9..5c68c160bf 100644 --- a/openmeter/ledger/breakage/service.go +++ b/openmeter/ledger/breakage/service.go @@ -146,7 +146,7 @@ func (i PlanIssuanceInput) Validate() error { } if immediateReleaseAmount.GreaterThan(i.Amount) { - errs = append(errs, errors.New("immediate release amount cannot exceed amount")) + errs = append(errs, models.NewError(ledger.ErrCodeBreakageFailed, "immediate release amount cannot exceed amount", models.WithAttribute("reason", "ledger_breakage_release_exceeds_issuance"))) } if err := i.Currency.Validate(); err != nil { @@ -234,7 +234,7 @@ func (i ReleasePlanInput) Validate() error { } if i.Amount.GreaterThan(i.Plan.OpenAmount) { - errs = append(errs, errors.New("release amount cannot exceed open plan amount")) + errs = append(errs, models.NewError(ledger.ErrCodeBreakageFailed, "release amount cannot exceed open plan amount", models.WithAttribute("reason", "ledger_breakage_release_exceeds_open_amount"))) } if i.Plan.FBOAddress == nil { @@ -291,7 +291,7 @@ func (i ReopenReleaseInput) Validate() error { } if i.Amount.GreaterThan(i.Release.OpenAmount) { - errs = append(errs, errors.New("reopen amount cannot exceed open release amount")) + errs = append(errs, models.NewError(ledger.ErrCodeBreakageFailed, "reopen amount cannot exceed open release amount", models.WithAttribute("reason", "ledger_breakage_reopen_exceeds_open_amount"))) } if i.Release.FBOAddress == nil { @@ -654,7 +654,7 @@ func (s *service) ListExpiredRecords(ctx context.Context, input ListExpiredRecor // Deferred ledger FKs reject discarded postings at the enclosing transaction's commit. func (s *service) recordPosting(ctx context.Context, input PostingInput, record Record, posting ledger.TransactionInput) (ledger.TransactionInput, error) { if _, err := transaction.GetDriverFromContext(ctx); err != nil { - return nil, fmt.Errorf("breakage requires the caller's transaction: %w", err) + return nil, models.NewError(ledger.ErrCodeBreakageFailed, fmt.Sprintf("breakage requires the caller's transaction: %v", err), models.WithAttribute("reason", "ledger_breakage_transaction_required"), models.WithCause(err)) } posting, err := ledger.PreassignIDs(posting) @@ -777,7 +777,7 @@ func (s *service) resolveBreakageTemplate( } if len(inputs) != 1 { - return nil, fmt.Errorf("expected one breakage transaction input, got %d", len(inputs)) + return nil, models.NewError(ledger.ErrCodeBreakageFailed, fmt.Sprintf("expected one breakage transaction input, got %d", len(inputs)), models.WithAttribute("reason", "ledger_breakage_transaction_count_invalid")) } kind, err := breakageKindForTemplate(template) diff --git a/openmeter/ledger/breakage/service_test.go b/openmeter/ledger/breakage/service_test.go index 8bafd04348..bbff41e627 100644 --- a/openmeter/ledger/breakage/service_test.go +++ b/openmeter/ledger/breakage/service_test.go @@ -240,6 +240,11 @@ func TestBreakageRequiresCallerTransaction(t *testing.T) { inputs, err := service.PlanIssuance(t.Context(), expiringIssuance(env, ulid.Make().String())) require.ErrorContains(t, err, "requires the caller's transaction") + + var diagnostic models.Error + require.ErrorAs(t, err, &diagnostic) + require.Equal(t, "ledger_breakage_transaction_required", diagnostic.Attributes()["reason"]) + require.Nil(t, inputs) requireNoBreakageRows(t, env) } @@ -295,3 +300,40 @@ func requireNoBreakageRows(t *testing.T, env *ledgertestutils.IntegrationEnv) { require.NoError(t, err) require.Zero(t, count) } + +func TestBreakageReleaseLimitReportsReason(t *testing.T) { + env, service := newBreakageEnv(t) + + // given: a plan for ten credits, created in the caller's transaction. + err := transaction.RunWithNoValue(t.Context(), enttx.NewCreator(env.DB), func(ctx context.Context) error { + _, err := service.PlanIssuance(ctx, expiringIssuance(env, ulid.Make().String())) + require.NoError(t, err) + + plans, err := service.ListPlans(ctx, breakage.ListPlansInput{ + CustomerID: env.CustomerID, + Currency: env.Currency, + AsOf: env.Now(), + }) + require.NoError(t, err) + require.Len(t, plans, 1) + + // when: consumption tries to release more than the plan holds. + _, err = service.ReleasePlan(ctx, breakage.ReleasePlanInput{ + PostingInput: breakage.PostingInput{TransactionGroupID: ulid.Make().String()}, + Plan: plans[0], + Amount: alpacadecimal.NewFromInt(11), + SourceKind: breakage.SourceKindUsage, + }) + + // then: callers can classify the rejection without parsing its message. + require.ErrorContains(t, err, "release amount cannot exceed open plan amount") + + var diagnostic models.Error + require.ErrorAs(t, err, &diagnostic) + require.Equal(t, "ledger_breakage_release_exceeds_open_amount", diagnostic.Attributes()["reason"]) + + return err + }) + require.Error(t, err) + requireNoBreakageRows(t, env) +} diff --git a/openmeter/ledger/chargeadapter/flatfee_test.go b/openmeter/ledger/chargeadapter/flatfee_test.go index 512bae8e67..760eb2a779 100644 --- a/openmeter/ledger/chargeadapter/flatfee_test.go +++ b/openmeter/ledger/chargeadapter/flatfee_test.go @@ -8,6 +8,7 @@ import ( "github.com/alpacahq/alpacadecimal" "github.com/oklog/ulid/v2" "github.com/stretchr/testify/require" + metricnoop "go.opentelemetry.io/otel/metric/noop" "github.com/openmeterio/openmeter/openmeter/billing" chargeflatfee "github.com/openmeterio/openmeter/openmeter/billing/charges/flatfee" @@ -721,6 +722,7 @@ func newFlatFeeHandlerTestEnv(t *testing.T) *flatFeeHandlerTestEnv { advanceService := advancetestutils.NewService(t, base.Deps, breakageService) collectorService, err := ledgercollector.NewService(ledgercollector.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: omtestutils.NewDiscardLogger(t), Advance: advanceService, Ledger: base.Deps.HistoricalLedger, diff --git a/openmeter/ledger/chargeadapter/origin_test.go b/openmeter/ledger/chargeadapter/origin_test.go index e87c451e5e..d857efded2 100644 --- a/openmeter/ledger/chargeadapter/origin_test.go +++ b/openmeter/ledger/chargeadapter/origin_test.go @@ -13,6 +13,7 @@ import ( "github.com/samber/lo" "github.com/samber/mo" "github.com/stretchr/testify/require" + metricnoop "go.opentelemetry.io/otel/metric/noop" "github.com/openmeterio/openmeter/openmeter/billing/charges/creditpurchase" "github.com/openmeterio/openmeter/openmeter/billing/charges/legacylineage" @@ -59,6 +60,7 @@ func newOriginTestEnv(t *testing.T, custom bool) *originTestEnv { advanceService := advancetestutils.NewService(t, base.Deps, base.breakage) collect, err := collector.NewService(collector.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: omtestutils.NewDiscardLogger(t), Advance: advanceService, Ledger: base.Deps.HistoricalLedger, diff --git a/openmeter/ledger/chargeadapter/usagebased_test.go b/openmeter/ledger/chargeadapter/usagebased_test.go index 432729c0e1..e8148c173b 100644 --- a/openmeter/ledger/chargeadapter/usagebased_test.go +++ b/openmeter/ledger/chargeadapter/usagebased_test.go @@ -8,6 +8,7 @@ import ( "github.com/alpacahq/alpacadecimal" "github.com/oklog/ulid/v2" "github.com/stretchr/testify/require" + metricnoop "go.opentelemetry.io/otel/metric/noop" "github.com/openmeterio/openmeter/openmeter/billing" "github.com/openmeterio/openmeter/openmeter/billing/charges/legacylineage" @@ -579,6 +580,7 @@ func newUsageBasedHandlerTestEnv(t *testing.T) *usageBasedHandlerTestEnv { advanceService := advancetestutils.NewService(t, base.Deps, breakageService) collectorService, err := ledgercollector.NewService(ledgercollector.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: omtestutils.NewDiscardLogger(t), Advance: advanceService, Ledger: base.Deps.HistoricalLedger, diff --git a/openmeter/ledger/collector/collection_fbo_test.go b/openmeter/ledger/collector/collection_fbo_test.go index 866f9b2952..bb416839b9 100644 --- a/openmeter/ledger/collector/collection_fbo_test.go +++ b/openmeter/ledger/collector/collection_fbo_test.go @@ -10,6 +10,7 @@ import ( "github.com/oklog/ulid/v2" "github.com/samber/lo" "github.com/stretchr/testify/require" + metricnoop "go.opentelemetry.io/otel/metric/noop" "github.com/openmeterio/openmeter/openmeter/currencies" currencytestutils "github.com/openmeterio/openmeter/openmeter/currencies/testutils" @@ -561,6 +562,7 @@ func newTestAccrualCollector(t testing.TB, env *ledgertestutils.IntegrationEnv) t.Helper() breakageService := newTestBreakageService(t, env) service, err := fbo.NewService(fbo.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: omtestutils.NewDiscardLogger(t), Advance: advancetestutils.NewService(t, env.Deps, breakageService), Breakage: breakageService, diff --git a/openmeter/ledger/collector/correction/legacy_references.go b/openmeter/ledger/collector/correction/legacy_references.go index 034a75b652..76138a575b 100644 --- a/openmeter/ledger/collector/correction/legacy_references.go +++ b/openmeter/ledger/collector/correction/legacy_references.go @@ -10,6 +10,7 @@ import ( "github.com/openmeterio/openmeter/openmeter/ledger" "github.com/openmeterio/openmeter/openmeter/ledger/transactions" + "github.com/openmeterio/openmeter/pkg/models" ) // correctedSourceAmounts reads immutable correction links so a later partial @@ -76,7 +77,7 @@ func reserveCorrectionSources(entries []ledger.Entry, amount alpacadecimal.Decim } if remaining.IsPositive() { - return nil, fmt.Errorf("correction exceeds remaining source capacity by %s", remaining) + return nil, models.NewError(ledger.ErrCodeCorrectionFailed, fmt.Sprintf("correction exceeds remaining source capacity by %s", remaining), models.WithAttribute("reason", "ledger_correction_exceeds_source_capacity")) } return out, nil @@ -137,12 +138,12 @@ func (c *Corrector) recognizedSourceAmounts(ctx context.Context, input recognize } if !selected.Equal(input.amount) { - return nil, fmt.Errorf("recognition source coverage %s does not match correction amount %s", selected, input.amount) + return nil, models.NewError(ledger.ErrCodeCorrectionFailed, fmt.Sprintf("recognition source coverage %s does not match correction amount %s", selected, input.amount), models.WithAttribute("reason", "ledger_correction_recognition_coverage_mismatch")) } for _, remaining := range accrued { if !remaining.IsZero() { - return nil, fmt.Errorf("recognition correction leaves unmatched accrued amount %s", remaining) + return nil, models.NewError(ledger.ErrCodeCorrectionFailed, fmt.Sprintf("recognition correction leaves unmatched accrued amount %s", remaining), models.WithAttribute("reason", "ledger_correction_accrued_unmatched")) } } diff --git a/openmeter/ledger/collector/correction/origin_correction.go b/openmeter/ledger/collector/correction/origin_correction.go index 23ff6a86eb..9b09ec1351 100644 --- a/openmeter/ledger/collector/correction/origin_correction.go +++ b/openmeter/ledger/collector/correction/origin_correction.go @@ -29,11 +29,11 @@ func (c *Corrector) prepareOriginCorrection(ctx context.Context, input Input, co for _, entry := range entries { if lo.FromPtr(entry.Provenance().SpendChargeID) != input.ChargeID { - return correctionPlan{}, fmt.Errorf("collection origin belongs to a different spend charge") + return correctionPlan{}, models.NewError(ledger.ErrCodeCorrectionFailed, "collection origin belongs to a different spend charge", models.WithAttribute("reason", "ledger_correction_spend_charge_mismatch")) } if entry.Provenance().CollectionOriginID == nil { - return correctionPlan{}, fmt.Errorf("collection mixes tracked and legacy source entries") + return correctionPlan{}, models.NewError(ledger.ErrCodeCorrectionFailed, "collection mixes tracked and legacy source entries", models.WithAttribute("reason", "ledger_correction_mixed_source_tracking")) } } @@ -207,7 +207,7 @@ func (h originReferences) reverseEarnings(at time.Time, selections []correctionS } if remaining.IsPositive() { - return nil, fmt.Errorf("earnings position lacks original recognition references") + return nil, models.NewError(ledger.ErrCodeCorrectionFailed, "earnings position lacks original recognition references", models.WithAttribute("reason", "ledger_correction_recognition_reference_missing")) } } @@ -248,7 +248,7 @@ func (h originReferences) backfillCorrections(selections []correctionSelection) } if remaining.IsPositive() { - return nil, fmt.Errorf("funded position lacks original backing references") + return nil, models.NewError(ledger.ErrCodeCorrectionFailed, "funded position lacks original backing references", models.WithAttribute("reason", "ledger_correction_backing_reference_missing")) } } diff --git a/openmeter/ledger/collector/correction/origin_positions.go b/openmeter/ledger/collector/correction/origin_positions.go index 119b98c96e..95d29977ca 100644 --- a/openmeter/ledger/collector/correction/origin_positions.go +++ b/openmeter/ledger/collector/correction/origin_positions.go @@ -9,6 +9,7 @@ import ( "github.com/samber/mo" "github.com/openmeterio/openmeter/openmeter/ledger" + "github.com/openmeterio/openmeter/pkg/models" ) const unknownOriginSource = "uncovered" @@ -33,7 +34,7 @@ func (c *Corrector) readOriginPositions(ctx context.Context, input Input, id str for _, bucket := range buckets { if lo.FromPtr(bucket.GroupByValues[ledger.BalanceBucketGroupBySpendChargeID]) != input.ChargeID { - return nil, fmt.Errorf("collection origin belongs to a different spend charge") + return nil, models.NewError(ledger.ErrCodeCorrectionFailed, "collection origin belongs to a different spend charge", models.WithAttribute("reason", "ledger_correction_spend_charge_mismatch")) } currency := bucket.Address.Route().Route().Currency.IdentityKey() diff --git a/openmeter/ledger/collector/correction/origin_references.go b/openmeter/ledger/collector/correction/origin_references.go index 9d03112f4d..15cb76245d 100644 --- a/openmeter/ledger/collector/correction/origin_references.go +++ b/openmeter/ledger/collector/correction/origin_references.go @@ -14,6 +14,7 @@ import ( "github.com/openmeterio/openmeter/openmeter/ledger" "github.com/openmeterio/openmeter/openmeter/ledger/advance" "github.com/openmeterio/openmeter/openmeter/ledger/transactions" + "github.com/openmeterio/openmeter/pkg/models" ) // originPair retains immutable routes and exact reversal capacity for the writer. @@ -37,7 +38,7 @@ func (p originPair) correctionSource() advance.CorrectionSource { func (pair originPair) reverse(at time.Time, amount alpacadecimal.Decimal) (ledger.TransactionInput, error) { if amount.GreaterThan(pair.remaining) { - return nil, fmt.Errorf("reversal exceeds remaining original entry amount") + return nil, models.NewError(ledger.ErrCodeCorrectionFailed, "reversal exceeds remaining original entry amount", models.WithAttribute("reason", "ledger_correction_exceeds_original_entry")) } return transactions.ReverseOriginEntryPair(transactions.ReverseOriginEntryPairInput{ @@ -97,14 +98,14 @@ func (h originReferences) attributionForBackfill(backfill *originPair) (*originP } if attribution != nil { - return nil, fmt.Errorf("ambiguous advance attribution for origin") + return nil, models.NewError(ledger.ErrCodeCorrectionFailed, "ambiguous advance attribution for origin", models.WithAttribute("reason", "ledger_correction_advance_attribution_ambiguous")) } attribution = pair } if attribution == nil { - return nil, fmt.Errorf("advance backfill has no matching receivable attribution") + return nil, models.NewError(ledger.ErrCodeCorrectionFailed, "advance backfill has no matching receivable attribution", models.WithAttribute("reason", "ledger_correction_receivable_attribution_missing")) } return attribution, nil @@ -237,14 +238,14 @@ func (c *Corrector) loadOriginReferences(ctx context.Context, namespace, collect for _, key := range order { pair := pairs[key] if pair.negativeEntry == nil || pair.positiveEntry == nil || !pair.negativeEntry.Amount().Neg().Equal(pair.positiveEntry.Amount()) { - return originReferences{}, fmt.Errorf("unbalanced origin pair in transaction %s", tx.ID().ID) + return originReferences{}, models.NewError(ledger.ErrCodeCorrectionFailed, fmt.Sprintf("unbalanced origin pair in transaction %s", tx.ID().ID), models.WithAttribute("reason", "ledger_correction_origin_pair_unbalanced")) } negativeRemaining := pair.negativeEntry.Amount().Add(corrections[pair.negativeEntry.ID().ID]).Neg() positiveRemaining := pair.positiveEntry.Amount().Add(corrections[pair.positiveEntry.ID().ID]) if negativeRemaining.IsNegative() || !negativeRemaining.Equal(positiveRemaining) { - return originReferences{}, fmt.Errorf("invalid prior reversals in origin transaction %s", tx.ID().ID) + return originReferences{}, models.NewError(ledger.ErrCodeCorrectionFailed, fmt.Sprintf("invalid prior reversals in origin transaction %s", tx.ID().ID), models.WithAttribute("reason", "ledger_correction_prior_reversals_invalid")) } role, err := originPairRole(pair.negativeEntry, pair.positiveEntry) @@ -277,7 +278,7 @@ func (h originReferences) pairForTransaction(id string) (*originPair, error) { } if found == nil { - return nil, fmt.Errorf("original transaction %s missing from origin history", id) + return nil, models.NewError(ledger.ErrCodeCorrectionFailed, fmt.Sprintf("original transaction %s missing from origin history", id), models.WithAttribute("reason", "ledger_correction_original_transaction_missing")) } return found, nil diff --git a/openmeter/ledger/collector/correction/plan.go b/openmeter/ledger/collector/correction/plan.go index 02334efa6e..9c3d5cff7f 100644 --- a/openmeter/ledger/collector/correction/plan.go +++ b/openmeter/ledger/collector/correction/plan.go @@ -9,6 +9,7 @@ import ( "github.com/alpacahq/alpacadecimal" + "github.com/openmeterio/openmeter/openmeter/ledger" "github.com/openmeterio/openmeter/pkg/models" ) @@ -119,7 +120,7 @@ func planCollectionCorrection(input collectionCorrectionInput) ([]correctionSele } if remaining.IsPositive() { - return nil, fmt.Errorf("correction exceeds remaining collection balance by %s", remaining) + return nil, models.NewError(ledger.ErrCodeCorrectionFailed, fmt.Sprintf("correction exceeds remaining collection balance by %s", remaining), models.WithAttribute("reason", "ledger_correction_exceeds_remaining_balance")) } return out, nil diff --git a/openmeter/ledger/collector/correction/plan_test.go b/openmeter/ledger/collector/correction/plan_test.go index 829ec20205..122a499e60 100644 --- a/openmeter/ledger/collector/correction/plan_test.go +++ b/openmeter/ledger/collector/correction/plan_test.go @@ -6,6 +6,8 @@ import ( "github.com/alpacahq/alpacadecimal" "github.com/stretchr/testify/require" + + "github.com/openmeterio/openmeter/pkg/models" ) func TestCollectionCorrectionSelectsSourceBeforeState(t *testing.T) { @@ -79,4 +81,8 @@ func TestCollectionCorrectionUsesOriginalBackingOrder(t *testing.T) { amount: alpacadecimal.NewFromInt(16), }) require.ErrorContains(t, err, "exceeds remaining collection balance") + + var diagnostic models.Error + require.ErrorAs(t, err, &diagnostic) + require.Equal(t, "ledger_correction_exceeds_remaining_balance", diagnostic.Attributes()["reason"]) } diff --git a/openmeter/ledger/collector/service.go b/openmeter/ledger/collector/service.go index e6cf549686..1b431641f1 100644 --- a/openmeter/ledger/collector/service.go +++ b/openmeter/ledger/collector/service.go @@ -8,6 +8,7 @@ import ( "time" "github.com/alpacahq/alpacadecimal" + "go.opentelemetry.io/otel/metric" "github.com/openmeterio/openmeter/openmeter/billing/charges/models/creditrealization" "github.com/openmeterio/openmeter/openmeter/currencies" @@ -33,6 +34,7 @@ type Service interface { type Config struct { Logger *slog.Logger + Meter metric.Meter Advance advance.Service Ledger ledger.Ledger Dependencies transactions.ResolverDependencies @@ -50,6 +52,10 @@ func (c Config) Validate() error { errs = append(errs, errors.New("logger is required")) } + if c.Meter == nil { + errs = append(errs, errors.New("meter is required")) + } + if c.Advance == nil { errs = append(errs, errors.New("advance service is required")) } @@ -213,6 +219,7 @@ func NewService(config Config) (Service, error) { } fboService, err := fbo.NewService(fbo.Config{ + Meter: config.Meter, Logger: config.Logger, Advance: config.Advance, Dependencies: config.Dependencies, diff --git a/openmeter/ledger/creditvoid/service.go b/openmeter/ledger/creditvoid/service.go index d2d8f3e263..fc56c84226 100644 --- a/openmeter/ledger/creditvoid/service.go +++ b/openmeter/ledger/creditvoid/service.go @@ -302,7 +302,7 @@ func (s *service) planVoid(ctx context.Context, input VoidCreditPurchaseInput) ( expiryPlan, ok := openPlansBySubAccount[bucket.Address.SubAccountID()] if ok { if expiryPlan.OpenAmount.LessThan(slice.amount) { - return voidPlan{}, fmt.Errorf("open expiry breakage plan %s amount %s is less than voided amount %s", expiryPlan.ID.ID, expiryPlan.OpenAmount, slice.amount) + return voidPlan{}, models.NewError(ledger.ErrCodeVoidFailed, fmt.Sprintf("open expiry breakage plan %s amount %s is less than voided amount %s", expiryPlan.ID.ID, expiryPlan.OpenAmount, slice.amount), models.WithAttribute("reason", "ledger_void_exceeds_open_breakage")) } slice.expiryPlan = &expiryPlan @@ -423,7 +423,7 @@ func (s *service) resolveVoidSlice(ctx context.Context, input VoidCreditPurchase } if len(inputs) != 1 { - return nil, pendingVoidRecord{}, fmt.Errorf("expected one issue correction transaction input, got %d", len(inputs)) + return nil, pendingVoidRecord{}, models.NewError(ledger.ErrCodeVoidFailed, fmt.Sprintf("expected one issue correction transaction input, got %d", len(inputs)), models.WithAttribute("reason", "ledger_void_transaction_count_invalid")) } correctedFBO, correctedReceivable, err := correctionEntrySubAccounts(inputs[0]) @@ -432,7 +432,7 @@ func (s *service) resolveVoidSlice(ctx context.Context, input VoidCreditPurchase } if correctedFBO != slice.fboAddress.SubAccountID() { - return nil, pendingVoidRecord{}, fmt.Errorf("issue correction FBO sub-account %s does not match voided FBO sub-account %s", correctedFBO, slice.fboAddress.SubAccountID()) + return nil, pendingVoidRecord{}, models.NewError(ledger.ErrCodeVoidFailed, fmt.Sprintf("issue correction FBO sub-account %s does not match voided FBO sub-account %s", correctedFBO, slice.fboAddress.SubAccountID()), models.WithAttribute("reason", "ledger_void_source_account_mismatch")) } record := pendingVoidRecord{ @@ -488,7 +488,7 @@ func (s *service) originalIssueTransaction(ctx context.Context, input VoidCredit } if page.NextCursor == nil { - return nil, fmt.Errorf("issue transaction for charge %s FBO sub-account %s not found", input.ChargeID, slice.fboAddress.SubAccountID()) + return nil, models.NewError(ledger.ErrCodeVoidFailed, fmt.Sprintf("issue transaction for charge %s FBO sub-account %s not found", input.ChargeID, slice.fboAddress.SubAccountID()), models.WithAttribute("reason", "ledger_void_original_transaction_missing")) } cursor = page.NextCursor diff --git a/openmeter/ledger/customerbalance/testenv_test.go b/openmeter/ledger/customerbalance/testenv_test.go index 17a97e0cd7..6a2bb11ec2 100644 --- a/openmeter/ledger/customerbalance/testenv_test.go +++ b/openmeter/ledger/customerbalance/testenv_test.go @@ -9,6 +9,7 @@ import ( "github.com/alpacahq/alpacadecimal" "github.com/samber/lo" "github.com/stretchr/testify/require" + metricnoop "go.opentelemetry.io/otel/metric/noop" "github.com/openmeterio/openmeter/openmeter/app" "github.com/openmeterio/openmeter/openmeter/billing" @@ -251,6 +252,7 @@ func newTestEnv(t *testing.T) *testEnv { advanceService := advancetestutils.NewService(t, base.Deps, breakageService) collectorService, err := ledgercollector.NewService(ledgercollector.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: logger, Advance: advanceService, Ledger: base.Deps.HistoricalLedger, diff --git a/openmeter/ledger/errors.go b/openmeter/ledger/errors.go index 77b1a14c00..f422d2505a 100644 --- a/openmeter/ledger/errors.go +++ b/openmeter/ledger/errors.go @@ -143,3 +143,13 @@ var ErrTaxBehaviorInvalid = models.NewValidationIssue( ErrCodeTaxBehaviorInvalid, "ledger tax behavior is invalid", ) + +const ( + ErrCodeAdvanceFailed models.ErrorCode = "ledger_advance_failed" + ErrCodeBreakageFailed models.ErrorCode = "ledger_breakage_failed" + ErrCodeCorrectionFailed models.ErrorCode = "ledger_correction_failed" + ErrCodeFBOConsumptionFailed models.ErrorCode = "ledger_fbo_consumption_failed" + ErrCodeRecognitionFailed models.ErrorCode = "ledger_recognition_failed" + ErrCodeReversalInvalid models.ErrorCode = "ledger_reversal_invalid" + ErrCodeVoidFailed models.ErrorCode = "ledger_void_failed" +) diff --git a/openmeter/ledger/fbo/README.md b/openmeter/ledger/fbo/README.md index 12a6e33fc4..b791cd1eaf 100644 --- a/openmeter/ledger/fbo/README.md +++ b/openmeter/ledger/fbo/README.md @@ -18,3 +18,9 @@ source selection and consumption. Receivable coverage never creates advance. reducing any later recorded balance below zero. It takes the minimum per source before combining buckets. Future breakage is excluded because consumption releases it; breakage at or before `AsOf` remains part of the opening balance. + +Negative source minima are excluded from spendable credit and increment +`openmeter.ledger.fbo.negative_source_balance.observations`. This counts source +observations, including repeated reads, not distinct sources or committed errors. +Each observation also logs a warning with the source, query boundary, and minimum +balance. diff --git a/openmeter/ledger/fbo/balance.go b/openmeter/ledger/fbo/balance.go index 74fb67ac58..a2b7715a4c 100644 --- a/openmeter/ledger/fbo/balance.go +++ b/openmeter/ledger/fbo/balance.go @@ -77,6 +77,10 @@ func (s *service) GetSpendableBalanceBuckets(ctx context.Context, query Spendabl spendable := make([]ledger.BalanceBucket, 0, len(buckets)) bySubaccount := make(map[string]int) for _, bucket := range buckets { + if bucket.SettledAmount.IsNegative() { + s.reportNegativeSourceBalance(ctx, query, bucket) + } + if !bucket.SettledAmount.IsPositive() { continue } @@ -103,3 +107,17 @@ func (s *service) GetSpendableBalanceBuckets(ctx context.Context, query Spendabl return spendable, nil } + +// Repeated reads count again: these are observations, not distinct broken sources. +func (s *service) reportNegativeSourceBalance(ctx context.Context, query SpendableBalanceBucketQuery, bucket ledger.BalanceBucket) { + s.negativeSourceObservations.Add(ctx, 1) + + s.logger.WarnContext(ctx, "negative FBO source minimum balance observed", + "namespace", query.Namespace, + "account_id", *query.Filters.AccountID, + "sub_account_id", bucket.Address.SubAccountID(), + "source_charge_id", bucket.GroupByValues[ledger.BalanceBucketGroupBySourceChargeID], + "as_of", *query.Filters.AsOf, + "minimum_balance", bucket.SettledAmount.String(), + ) +} diff --git a/openmeter/ledger/fbo/balance_test.go b/openmeter/ledger/fbo/balance_test.go index f75bb663c1..e24b1570cc 100644 --- a/openmeter/ledger/fbo/balance_test.go +++ b/openmeter/ledger/fbo/balance_test.go @@ -1,7 +1,10 @@ package fbo_test import ( + "bytes" "context" + "encoding/json" + "log/slog" "testing" "time" @@ -10,9 +13,12 @@ import ( "github.com/samber/lo" "github.com/samber/mo" "github.com/stretchr/testify/require" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" enttx "github.com/openmeterio/openmeter/openmeter/ent/tx" "github.com/openmeterio/openmeter/openmeter/ledger" + advancetestutils "github.com/openmeterio/openmeter/openmeter/ledger/advance/testutils" "github.com/openmeterio/openmeter/openmeter/ledger/fbo" ledgertestutils "github.com/openmeterio/openmeter/openmeter/ledger/testutils" "github.com/openmeterio/openmeter/openmeter/ledger/transactions" @@ -240,3 +246,120 @@ func consumeAndCommit(t *testing.T, env *ledgertestutils.IntegrationEnv, service return group } + +func TestSpendableBalanceReportsNegativeSources(t *testing.T) { + env, _, breakageService := newFBOEnv(t) + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + t.Cleanup(func() { require.NoError(t, provider.Shutdown(t.Context())) }) + + var logs bytes.Buffer + service, err := fbo.NewService(fbo.Config{ + Logger: slog.New(slog.NewJSONHandler(&logs, nil)), + Meter: provider.Meter("fbo-test"), + Dependencies: transactions.ResolverDependencies{ + AccountService: env.Deps.ResolversService, + AccountCatalog: env.Deps.AccountService, + BalanceQuerier: env.Deps.HistoricalLedger, + }, + Advance: advancetestutils.NewService(t, env.Deps, breakageService), + Breakage: breakageService, + AccountLocker: env.Deps.AccountService, + }) + require.NoError(t, err) + + // given: one exhausted source and one with four credits left. + exhaustedSource, availableSource := ulid.Make().String(), ulid.Make().String() + for _, grant := range []struct { + source string + amount int64 + }{ + {exhaustedSource, 10}, {availableSource, 4}, + } { + bookFBOTransactions(t, env, transactions.IssueCustomerReceivableTemplate{ + At: env.Now(), Amount: alpacadecimal.NewFromInt(grant.amount), + Currency: env.CurrencyReference(), SourceChargeID: &grant.source, + }) + } + + address := env.FBOSubAccount(t, ledger.DefaultCustomerFBOPriority).Address() + bookFBOTransactions(t, env, transactions.TransferCustomerFBOToAccruedTemplate{ + At: env.Now(), Currency: env.CurrencyReference(), + Sources: []transactions.PostingAmount{{ + Address: address, Amount: alpacadecimal.NewFromInt(10), + Identity: ledger.EntryIdentityParts{Provenance: ledger.Provenance{ + SourceChargeID: &exhaustedSource, SpendChargeID: lo.ToPtr("usage"), + }}, + }}, + }) + query := fbo.SpendableBalanceBucketQuery{ + Namespace: env.Namespace, + Filters: ledger.Filters{ + AccountID: lo.ToPtr(env.CustomerAccounts.FBOAccount.ID().ID), + AsOf: lo.ToPtr(env.Now()), + Route: ledger.RouteFilter{Currency: env.CurrencyReference()}, + }, + } + + // when: reading healthy and zero balances, neither should be reported. + buckets, err := service.GetSpendableBalanceBuckets(t.Context(), query) + require.NoError(t, err) + require.Len(t, buckets, 1) + require.Equal(t, 4.0, buckets[0].SettledAmount.InexactFloat64()) + require.Empty(t, logs.String()) + var before metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &before)) + require.Empty(t, before.ScopeMetrics) + + // given: an extra future posting overspends the exhausted source by two. + bookFBOTransactions(t, env, transactions.TransferCustomerFBOToAccruedTemplate{ + At: env.Now().Add(time.Hour), Currency: env.CurrencyReference(), + Sources: []transactions.PostingAmount{{ + Address: address, Amount: alpacadecimal.NewFromInt(2), + Identity: ledger.EntryIdentityParts{Provenance: ledger.Provenance{ + SourceChargeID: &exhaustedSource, SpendChargeID: lo.ToPtr("extra-usage"), + }}, + }}, + }) + + // when: repeated reads observe that negative source, with either grouping. + buckets, err = service.GetSpendableBalanceBuckets(t.Context(), query) + require.NoError(t, err) + require.Len(t, buckets, 1) + require.Equal(t, 4.0, buckets[0].SettledAmount.InexactFloat64()) + require.Empty(t, buckets[0].GroupByValues) + + query.GroupBy = []string{ledger.BalanceBucketGroupBySourceChargeID} + buckets, err = service.GetSpendableBalanceBuckets(t.Context(), query) + require.NoError(t, err) + require.Len(t, buckets, 1) + require.Equal(t, 4.0, buckets[0].SettledAmount.InexactFloat64()) + require.Equal(t, availableSource, *buckets[0].GroupByValues[ledger.BalanceBucketGroupBySourceChargeID]) + + // then: every observation increments the counter and logs its source. + var metrics metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &metrics)) + require.Len(t, metrics.ScopeMetrics, 1) + require.Len(t, metrics.ScopeMetrics[0].Metrics, 1) + observed := metrics.ScopeMetrics[0].Metrics[0] + require.Equal(t, "openmeter.ledger.fbo.negative_source_balance.observations", observed.Name) + counter, ok := observed.Data.(metricdata.Sum[int64]) + require.True(t, ok) + require.Len(t, counter.DataPoints, 1) + require.Equal(t, int64(2), counter.DataPoints[0].Value) + require.Zero(t, counter.DataPoints[0].Attributes.Len()) + + logLines := bytes.Split(bytes.TrimSpace(logs.Bytes()), []byte("\n")) + require.Len(t, logLines, 2) + for _, line := range logLines { + var diagnostic map[string]any + require.NoError(t, json.Unmarshal(line, &diagnostic)) + require.Equal(t, "WARN", diagnostic["level"]) + require.Equal(t, env.Namespace, diagnostic["namespace"]) + require.Equal(t, env.CustomerAccounts.FBOAccount.ID().ID, diagnostic["account_id"]) + require.Equal(t, address.SubAccountID(), diagnostic["sub_account_id"]) + require.Equal(t, exhaustedSource, diagnostic["source_charge_id"]) + require.Equal(t, "-2", diagnostic["minimum_balance"]) + require.Equal(t, env.Now().Format(time.RFC3339Nano), diagnostic["as_of"]) + } +} diff --git a/openmeter/ledger/fbo/consume.go b/openmeter/ledger/fbo/consume.go index 431464fa22..8752740e31 100644 --- a/openmeter/ledger/fbo/consume.go +++ b/openmeter/ledger/fbo/consume.go @@ -281,7 +281,7 @@ func assignSourceDebitIDs(creditTransactions []ledger.TransactionInput) (map[str originID := lo.FromPtr(entry.Provenance().CollectionOriginID) if _, exists := sourceDebitsByOrigin[originID]; exists { - return nil, fmt.Errorf("collection origin %s has multiple FBO source entries", originID) + return nil, models.NewError(ledger.ErrCodeFBOConsumptionFailed, fmt.Sprintf("collection origin %s has multiple FBO source entries", originID), models.WithAttribute("reason", "ledger_fbo_source_debit_ambiguous")) } sourceDebitsByOrigin[originID] = sourcePostingIDs{ @@ -328,7 +328,7 @@ func (s *service) releaseConsumedBreakage(ctx context.Context, input releaseCons for _, selection := range input.SelectedCredits { consumedSource, ok := input.SourceDebitsByOrigin[lo.FromPtr(selection.collectionOriginID)] if !ok { - return nil, errors.New("selected credit has no resolved FBO source posting") + return nil, models.NewError(ledger.ErrCodeFBOConsumptionFailed, "selected credit has no resolved FBO source posting", models.WithAttribute("reason", "ledger_fbo_source_debit_missing")) } if selection.source.breakagePlan != nil { @@ -341,7 +341,7 @@ func (s *service) releaseConsumedBreakage(ctx context.Context, input releaseCons // Legacy source-less plans can span several selected source slices; // cap their combined release before writing more bookkeeping. if selection.amount.GreaterThan(remaining) { - return nil, fmt.Errorf("release amount exceeds remaining breakage plan amount for %s", breakagePlan.ID.ID) + return nil, models.NewError(ledger.ErrCodeFBOConsumptionFailed, fmt.Sprintf("release amount exceeds remaining breakage plan amount for %s", breakagePlan.ID.ID), models.WithAttribute("reason", "ledger_breakage_release_exceeds_open_amount")) } remainingBreakageByPlanID[breakagePlan.ID.ID] = remaining.Sub(selection.amount) diff --git a/openmeter/ledger/fbo/consume_test.go b/openmeter/ledger/fbo/consume_test.go index 9c1f796048..f0c982553f 100644 --- a/openmeter/ledger/fbo/consume_test.go +++ b/openmeter/ledger/fbo/consume_test.go @@ -10,6 +10,7 @@ import ( "github.com/oklog/ulid/v2" "github.com/samber/lo" "github.com/stretchr/testify/require" + metricnoop "go.opentelemetry.io/otel/metric/noop" currencytestutils "github.com/openmeterio/openmeter/openmeter/currencies/testutils" enttx "github.com/openmeterio/openmeter/openmeter/ent/tx" @@ -167,6 +168,7 @@ func newFBOEnv(t *testing.T) (*ledgertestutils.IntegrationEnv, fbo.Service, brea breakageService, err := breakage.NewService(breakage.Config{Adapter: adapter, Dependencies: dependencies}) require.NoError(t, err) service, err := fbo.NewService(fbo.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: omtestutils.NewDiscardLogger(t), Dependencies: dependencies, Advance: advancetestutils.NewService(t, env.Deps, breakageService), diff --git a/openmeter/ledger/fbo/service.go b/openmeter/ledger/fbo/service.go index ef83afe6e2..f55b75b14d 100644 --- a/openmeter/ledger/fbo/service.go +++ b/openmeter/ledger/fbo/service.go @@ -3,8 +3,11 @@ package fbo import ( "context" "errors" + "fmt" "log/slog" + "go.opentelemetry.io/otel/metric" + "github.com/openmeterio/openmeter/openmeter/ledger" "github.com/openmeterio/openmeter/openmeter/ledger/advance" "github.com/openmeterio/openmeter/openmeter/ledger/breakage" @@ -20,6 +23,7 @@ type Service interface { type Config struct { Logger *slog.Logger + Meter metric.Meter Dependencies transactions.ResolverDependencies Advance advance.Service Breakage breakage.Service @@ -34,6 +38,10 @@ func (c Config) Validate() error { errs = append(errs, errors.New("logger is required")) } + if c.Meter == nil { + errs = append(errs, errors.New("meter is required")) + } + if c.Dependencies.AccountService == nil { errs = append(errs, errors.New("account service is required")) } @@ -62,10 +70,12 @@ func (c Config) Validate() error { } type service struct { - dependencies transactions.ResolverDependencies - advance advance.Service - breakage breakage.Service - accountLocker ledger.AccountLocker + logger *slog.Logger + negativeSourceObservations metric.Int64Counter + dependencies transactions.ResolverDependencies + advance advance.Service + breakage breakage.Service + accountLocker ledger.AccountLocker } var _ Service = (*service)(nil) @@ -75,10 +85,21 @@ func NewService(config Config) (Service, error) { return nil, err } + negativeSourceObservations, err := config.Meter.Int64Counter( + "openmeter.ledger.fbo.negative_source_balance.observations", + metric.WithDescription("Negative source minimum balances observed during spendable balance reads; repeated reads count again"), + metric.WithUnit("{observation}"), + ) + if err != nil { + return nil, fmt.Errorf("create negative source balance counter: %w", err) + } + return &service{ - dependencies: config.Dependencies, - advance: config.Advance, - breakage: config.Breakage, - accountLocker: config.AccountLocker, + logger: config.Logger, + negativeSourceObservations: negativeSourceObservations, + dependencies: config.Dependencies, + advance: config.Advance, + breakage: config.Breakage, + accountLocker: config.AccountLocker, }, nil } diff --git a/openmeter/ledger/recognizer/sources.go b/openmeter/ledger/recognizer/sources.go index 852c7ea19f..6af90b2331 100644 --- a/openmeter/ledger/recognizer/sources.go +++ b/openmeter/ledger/recognizer/sources.go @@ -129,7 +129,7 @@ func (s *service) planRecognition(ctx context.Context, in RecognizeEarningsInput func (s *service) recognitionGroup(ctx context.Context, namespace, id string, groups map[string]ledger.TransactionGroup) (ledger.TransactionGroup, error) { if id == "" { - return nil, fmt.Errorf("recognition lineage is missing its allocation or backing transaction group") + return nil, models.NewError(ledger.ErrCodeRecognitionFailed, "recognition lineage is missing its allocation or backing transaction group", models.WithAttribute("reason", "ledger_recognition_lineage_reference_missing")) } if group, ok := groups[id]; ok { @@ -211,7 +211,7 @@ func allocationAccruedSources(group ledger.TransactionGroup, sortHint int) ([]ac } } - return nil, fmt.Errorf("allocation sort hint %d out of range for recognition source group", sortHint) + return nil, models.NewError(ledger.ErrCodeRecognitionFailed, fmt.Sprintf("allocation sort hint %d out of range for recognition source group", sortHint), models.WithAttribute("reason", "ledger_recognition_allocation_reference_invalid")) } func sameFundingRoute(left, right ledger.Route) bool { diff --git a/openmeter/ledger/transactions/provenance.go b/openmeter/ledger/transactions/provenance.go index 85b0be55ce..f8015207dc 100644 --- a/openmeter/ledger/transactions/provenance.go +++ b/openmeter/ledger/transactions/provenance.go @@ -39,30 +39,30 @@ func (i ReverseOriginEntryPairInput) Validate() error { } if i.NegativeEntry.TransactionID() != i.Transaction.ID() || i.PositiveEntry.TransactionID() != i.Transaction.ID() { - errs = append(errs, errors.New("entries must belong to the original transaction")) + errs = append(errs, models.NewError(ledger.ErrCodeReversalInvalid, "entries must belong to the original transaction", models.WithAttribute("reason", "ledger_reversal_transaction_mismatch"))) } if i.NegativeEntry.Provenance().CollectionOriginID == nil || !equal.ComparablePtrEqual(i.NegativeEntry.Provenance().CollectionOriginID, i.PositiveEntry.Provenance().CollectionOriginID) { - errs = append(errs, errors.New("entries must share a collection origin")) + errs = append(errs, models.NewError(ledger.ErrCodeReversalInvalid, "entries must share a collection origin", models.WithAttribute("reason", "ledger_reversal_origin_mismatch"))) } if !i.NegativeEntry.Amount().IsNegative() || !i.PositiveEntry.Amount().IsPositive() || !i.NegativeEntry.Amount().Neg().Equal(i.PositiveEntry.Amount()) { - errs = append(errs, errors.New("original entries must form an equal and opposite pair")) + errs = append(errs, models.NewError(ledger.ErrCodeReversalInvalid, "original entries must form an equal and opposite pair", models.WithAttribute("reason", "ledger_reversal_pair_unbalanced"))) } if i.Amount.GreaterThan(i.PositiveEntry.Amount()) { - errs = append(errs, errors.New("amount exceeds original pair")) + errs = append(errs, models.NewError(ledger.ErrCodeReversalInvalid, "amount exceeds original pair", models.WithAttribute("reason", "ledger_reversal_exceeds_original_pair"))) } if !i.NegativeEntry.PostingAddress().Route().Route().Currency.Equal(i.PositiveEntry.PostingAddress().Route().Route().Currency) { - errs = append(errs, errors.New("entry-pair reversal cannot convert currency")) + errs = append(errs, models.NewError(ledger.ErrCodeReversalInvalid, "entry-pair reversal cannot convert currency", models.WithAttribute("reason", "ledger_reversal_currency_mismatch"))) } direction, err := ledger.TransactionDirectionFromAnnotations(i.Transaction.Annotations()) if err != nil { errs = append(errs, err) } else if direction != ledger.TransactionDirectionForward { - errs = append(errs, errors.New("only forward transactions can be reversed")) + errs = append(errs, models.NewError(ledger.ErrCodeReversalInvalid, "only forward transactions can be reversed", models.WithAttribute("reason", "ledger_reversal_direction_invalid"))) } return models.NewNillableGenericValidationError(errors.Join(errs...)) diff --git a/pkg/errorsx/diagnostics.go b/pkg/errorsx/diagnostics.go new file mode 100644 index 0000000000..49a9fb789d --- /dev/null +++ b/pkg/errorsx/diagnostics.go @@ -0,0 +1,72 @@ +package errorsx + +import ( + "log/slog" + "slices" + + "github.com/openmeterio/openmeter/pkg/models" +) + +// Keep all reasons from joined errors, but don't copy arbitrary error attributes +// into logs. They can contain request data or unbounded provider values. +func diagnosticAttributes(err error) []any { + // Bound the whole traversal so cycles and branching chains cannot recurse forever. + remaining := 1024 + var codes, reasons []string + appendUnique := func(values *[]string, value string) { + if value != "" && !slices.Contains(*values, value) { + *values = append(*values, value) + } + } + + var visit func(error) + visit = func(err error) { + if err == nil || remaining == 0 { + return + } + + remaining-- + + if issue, ok := err.(interface { + Code() models.ErrorCode + Attributes() models.Attributes + }); ok { + appendUnique(&codes, string(issue.Code())) + if reason, ok := issue.Attributes()["reason"].(string); ok { + appendUnique(&reasons, reason) + } + + // Skip earlier copies whose attributes were replaced, but keep the cause. + if caused, ok := err.(interface{ Cause() error }); ok { + visit(caused.Cause()) + } + + return + } + + switch wrapped := err.(type) { + case interface{ Unwrap() []error }: + for _, child := range wrapped.Unwrap() { + if remaining == 0 { + break + } + + visit(child) + } + case interface{ Unwrap() error }: + visit(wrapped.Unwrap()) + } + } + visit(err) + + var attrs []any + if len(codes) > 0 { + attrs = append(attrs, slog.Any("error_codes", codes)) + } + + if len(reasons) > 0 { + attrs = append(attrs, slog.Any("error_reasons", reasons)) + } + + return attrs +} diff --git a/pkg/errorsx/diagnostics_test.go b/pkg/errorsx/diagnostics_test.go new file mode 100644 index 0000000000..7f5f086ef0 --- /dev/null +++ b/pkg/errorsx/diagnostics_test.go @@ -0,0 +1,38 @@ +package errorsx + +import ( + "log/slog" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/openmeterio/openmeter/pkg/models" +) + +func TestDiagnosticAttributesBoundsCyclicErrors(t *testing.T) { + cycle := &cyclicDiagnosticError{} + err := models.NewError("ledger_failed", "accounting failed", models.WithCause(cycle)). + WithAttr("reason", "source_missing") + + attrs := diagnosticAttributes(err) + + require.Positive(t, cycle.visits) + require.LessOrEqual(t, cycle.visits, 1024) + require.Equal(t, []any{ + slog.Any("error_codes", []string{"ledger_failed"}), + slog.Any("error_reasons", []string{"source_missing"}), + }, attrs) +} + +type cyclicDiagnosticError struct { + visits int +} + +var _ error = (*cyclicDiagnosticError)(nil) + +func (*cyclicDiagnosticError) Error() string { return "cyclic error" } + +func (e *cyclicDiagnosticError) Unwrap() []error { + e.visits++ + return []error{e, e} +} diff --git a/pkg/errorsx/handler.go b/pkg/errorsx/handler.go index 53a925b467..b8c130717e 100644 --- a/pkg/errorsx/handler.go +++ b/pkg/errorsx/handler.go @@ -34,48 +34,52 @@ func NewSlogHandler(logger *slog.Logger) SlogHandler { } func (h SlogHandler) Handle(err error) { + attrs := diagnosticAttributes(err) + // Context canceled errors are logged as warnings. if contextx.IsCanceledError(err) { - h.Logger.Warn(err.Error()) + h.Logger.Warn(err.Error(), attrs...) return } if isAPIError(err) { - h.Logger.Warn(err.Error()) + h.Logger.Warn(err.Error(), attrs...) return } // Warn errors are logged as warnings. if wErr, ok := lo.ErrorsAs[*warnError](err); ok { - h.Logger.Warn(wErr.Error()) + h.Logger.Warn(wErr.Error(), attrs...) return } // All other errors are logged as errors. - h.Logger.Error(err.Error()) + h.Logger.Error(err.Error(), attrs...) } func (h SlogHandler) HandleContext(ctx context.Context, err error) { + attrs := diagnosticAttributes(err) + // Only downgrade cancellations of the caller's context. Independent dependency // cancellations remain failures while the caller is still waiting for a response. if errors.Is(ctx.Err(), context.Canceled) && contextx.IsCanceledError(err) { - h.Logger.WarnContext(ctx, err.Error()) + h.Logger.WarnContext(ctx, err.Error(), attrs...) return } if isAPIError(err) { - h.Logger.WarnContext(ctx, err.Error()) + h.Logger.WarnContext(ctx, err.Error(), attrs...) return } // Warn errors are logged as warnings. if wErr, ok := lo.ErrorsAs[*warnError](err); ok { - h.Logger.WarnContext(ctx, wErr.Error()) + h.Logger.WarnContext(ctx, wErr.Error(), attrs...) return } // All other errors are logged as errors. - h.Logger.ErrorContext(ctx, err.Error()) + h.Logger.ErrorContext(ctx, err.Error(), attrs...) } // NopHandler ignores all errors. Useful for testing. diff --git a/pkg/errorsx/handler_test.go b/pkg/errorsx/handler_test.go index 5fca3580c3..f592c5a16d 100644 --- a/pkg/errorsx/handler_test.go +++ b/pkg/errorsx/handler_test.go @@ -3,6 +3,7 @@ package errorsx import ( "bytes" "context" + "encoding/json" "errors" "fmt" "log/slog" @@ -11,6 +12,8 @@ import ( "github.com/stretchr/testify/require" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + + "github.com/openmeterio/openmeter/pkg/models" ) func TestSlogHandlerCancellation(t *testing.T) { @@ -50,3 +53,73 @@ func TestSlogHandlerWithoutContextRecognizesGRPCCancellation(t *testing.T) { handler.Handle(fmt.Errorf("authorize: %w", status.Error(codes.Canceled, "request canceled"))) require.Contains(t, output.String(), "level=WARN") } + +func TestSlogHandlerErrorDiagnostics(t *testing.T) { + base := models.NewValidationIssue("invalid_entry", "invalid entry").WithAttrs(models.Attributes{ + "reason": "old_reason", + }) + issue := base.WithAttrs(models.Attributes{ + "reason": "origin_mismatch", + "request_data": "must not become a log attribute", + }) + operational := models.NewError("ledger_collection_failed", "source exhausted", models.WithAttribute("reason", "ledger_source_exhausted")) + err := fmt.Errorf("collect: %w", errors.Join(issue, &issue, operational, operational, errors.New("unclassified"))) + + var output bytes.Buffer + handler := NewSlogHandler(slog.New(slog.NewJSONHandler(&output, nil))) + handler.HandleContext(t.Context(), err) + + var record map[string]any + require.NoError(t, json.Unmarshal(output.Bytes(), &record)) + require.Equal(t, err.Error(), record["msg"]) + require.Equal(t, "ERROR", record["level"]) + require.Equal(t, []any{"invalid_entry", "ledger_collection_failed"}, record["error_codes"]) + require.Equal(t, []any{"origin_mismatch", "ledger_source_exhausted"}, record["error_reasons"]) + require.NotContains(t, output.String(), "old_reason") + require.NotContains(t, output.String(), "request_data") + require.NotContains(t, output.String(), "must not become a log attribute") +} + +func TestSlogHandlerWarningDiagnostics(t *testing.T) { + err := NewWarnError(models.NewError("ledger_collection_failed", "source exhausted", models.WithAttribute("reason", "ledger_source_exhausted"))) + + var output bytes.Buffer + handler := NewSlogHandler(slog.New(slog.NewJSONHandler(&output, nil))) + handler.Handle(err) + + var record map[string]any + require.NoError(t, json.Unmarshal(output.Bytes(), &record)) + require.Equal(t, "WARN", record["level"]) + require.Equal(t, err.Error(), record["msg"]) + require.Equal(t, []any{"ledger_source_exhausted"}, record["error_reasons"]) + require.Equal(t, []any{"ledger_collection_failed"}, record["error_codes"]) +} + +func TestSlogHandlerUnclassifiedError(t *testing.T) { + var output bytes.Buffer + handler := NewSlogHandler(slog.New(slog.NewJSONHandler(&output, nil))) + handler.Handle(errors.New("unclassified failure")) + + var record map[string]any + require.NoError(t, json.Unmarshal(output.Bytes(), &record)) + require.Equal(t, "ERROR", record["level"]) + require.Equal(t, "unclassified failure", record["msg"]) + require.NotContains(t, record, "error_codes") + require.NotContains(t, record, "error_reasons") +} + +func TestSlogHandlerStructuredCauseDiagnostics(t *testing.T) { + cause := models.NewError("storage_failed", "storage unavailable").WithAttr("reason", "connection_lost") + err := models.NewError("ledger_failed", "cannot load source", models.WithCause(cause)). + WithAttr("reason", "old_reason").WithAttr("reason", "source_unavailable") + + var output bytes.Buffer + handler := NewSlogHandler(slog.New(slog.NewJSONHandler(&output, nil))) + handler.HandleContext(t.Context(), err) + + var record map[string]any + require.NoError(t, json.Unmarshal(output.Bytes(), &record)) + require.Equal(t, []any{"ledger_failed", "storage_failed"}, record["error_codes"]) + require.Equal(t, []any{"source_unavailable", "connection_lost"}, record["error_reasons"]) + require.NotContains(t, output.String(), "old_reason") +} diff --git a/pkg/framework/commonhttp/errors_test.go b/pkg/framework/commonhttp/errors_test.go index 34d6aaa497..97b1e23049 100644 --- a/pkg/framework/commonhttp/errors_test.go +++ b/pkg/framework/commonhttp/errors_test.go @@ -84,3 +84,22 @@ func TestGenericErrorEncoderStatusMapping(t *testing.T) { }) } } + +func TestStructuredSystemErrorIsNotHTTPValidation(t *testing.T) { + validation := models.NewValidationIssue("invalid_amount", "amount is invalid", commonhttp.WithHTTPStatusCodeAttribute(http.StatusBadRequest)) + systemErr := models.NewError("ledger_failed", "internal accounting failure", models.WithCause(validation)). + WithAttr("reason", "source_missing") + + response := httptest.NewRecorder() + require.False(t, commonhttp.GenericErrorEncoder()(t.Context(), systemErr, response, nil)) + commonhttp.ErrorEncoder(t.Context(), systemErr, response) + + require.Equal(t, http.StatusInternalServerError, response.Code) + require.NotContains(t, response.Body.String(), "source_missing") + require.NotContains(t, response.Body.String(), "internal accounting failure") + + // The same validation issue still uses its explicitly assigned HTTP status. + response = httptest.NewRecorder() + require.True(t, commonhttp.GenericErrorEncoder()(t.Context(), fmt.Errorf("validate: %w", validation), response, nil)) + require.Equal(t, http.StatusBadRequest, response.Code) +} diff --git a/pkg/models/structured_error.go b/pkg/models/structured_error.go new file mode 100644 index 0000000000..db0f7d65cc --- /dev/null +++ b/pkg/models/structured_error.go @@ -0,0 +1,358 @@ +package models + +import ( + "encoding/json" + "maps" + "reflect" + + "github.com/samber/lo" +) + +var ( + _ error = Error{} + _ json.Marshaler = Error{} + _ Equaler[Error] = Error{} +) + +type ErrorExtension map[string]any + +// Error carries diagnostic context without classifying the failure as validation. +type Error = structuredError[generalErrorKind] + +type generalErrorKind struct{} + +// The kind is only a type parameter: fluent methods keep the error's type. +type structuredError[K any] struct { + attributes Attributes + code ErrorCode + component ComponentName + message string + field *FieldDescriptor + severity ErrorSeverity + + // Earlier copies keep errors.Is(enriched, original) working after adding context. + previous error + // The underlying failure. Unwrap exposes it to errors.Is and errors.As. + cause error +} + +func (i structuredError[K]) MarshalJSON() ([]byte, error) { + m := i.AsErrorExtension() + + return json.Marshal(m) +} + +func (i structuredError[K]) Clone() structuredError[K] { + return structuredError[K]{ + attributes: i.attributes.Clone(), + code: i.code, + component: i.component, + message: i.message, + field: i.field, + severity: i.severity, + previous: i, + cause: i.cause, + } +} + +// Attributes returns a shallow-copy of the underlying attributes map. +func (i structuredError[K]) Attributes() Attributes { + return maps.Clone(i.attributes) +} + +func (i structuredError[K]) Code() ErrorCode { + return i.code +} + +func (i structuredError[K]) Component() ComponentName { + return i.component +} + +func (i structuredError[K]) Message() string { + return i.message +} + +func (i structuredError[K]) Field() *FieldDescriptor { + if i.field == nil { + return NewFieldSelectorGroup() + } + + return i.field +} + +func (i structuredError[K]) Severity() ErrorSeverity { + return i.severity +} + +func (i structuredError[K]) With(opts ...ErrorOption) structuredError[K] { + v := i.Clone() + for _, opt := range opts { + opt((*Error)(&v)) + } + + return v +} + +func (i structuredError[K]) WithField(parts ...*FieldDescriptor) structuredError[K] { + v := i.Clone() + group := NewFieldSelectorGroup(parts...) + + v.field = group + + return v +} + +func (i structuredError[K]) WithPathString(parts ...string) structuredError[K] { + return i.WithField(lo.Map(parts, func(item string, _ int) *FieldDescriptor { + return NewFieldSelector(item) + })...) +} + +func (i structuredError[K]) WithSeverity(s ErrorSeverity) structuredError[K] { + v := i.Clone() + v.severity = s + + return v +} + +func (i structuredError[K]) WithAttr(key any, value any) structuredError[K] { + if key == nil { + panic("validation issue attribute key must not be nil") + } + + if t := reflect.TypeOf(key); t == nil || !t.Comparable() { + panic("validation issue attribute key is not comparable") + } + + return i.WithAttrs(Attributes{ + key: value, + }) +} + +func (i structuredError[K]) WithAttrs(attrs Attributes) structuredError[K] { + if len(attrs) == 0 { + return i + } + + for k := range attrs { + if k == nil { + panic("validation issue attribute key must not be nil") + } + + if t := reflect.TypeOf(k); t == nil || !t.Comparable() { + panic("validation issue attribute key is not comparable") + } + } + + v := i.Clone() + v.attributes = i.attributes.Merge(attrs) + + return v +} + +func (i structuredError[K]) SetAttributes(attrs Attributes) structuredError[K] { + v := i.Clone() + + v.attributes = attrs + + return v +} + +func (i structuredError[K]) Error() string { + return i.message +} + +// Keep matching the original error after adding context. Earlier copies are +// identity history, not causes: Unwrap should expose the actual underlying failure. +func (i structuredError[K]) Is(target error) bool { + targetError, ok := asStructuredError[K](target) + if !ok { + return false + } + + for { + if i.Equal(targetError) { + return true + } + + previous, ok := asStructuredError[K](i.previous) + if !ok { + return false + } + + i = previous + } +} + +func (i structuredError[K]) Unwrap() error { return i.cause } + +// Cause returns the underlying failure, without diagnostic context added by this error. +func (i structuredError[K]) Cause() error { return i.cause } + +func (i structuredError[K]) Equal(other structuredError[K]) bool { + return maps.EqualFunc(i.attributes, other.attributes, equalErrorAttributeValue) && + i.code == other.code && + i.component == other.component && + i.message == other.message && + i.severity == other.severity && + i.Field().String() == other.Field().String() +} + +// equalErrorAttributeValue is used to compare attribute values for equality. +// It's better than reflect.DeepEqual because it handles nil values properly (DeepEqual would consider nil and empty value equal) +func equalErrorAttributeValue(left, right any) bool { + if left == nil || right == nil { + return left == right + } + + // Let's not use DeepEqual for comparable types + if t := reflect.TypeOf(left); t != nil && t.Comparable() { + return left == right + } + + return reflect.DeepEqual(left, right) +} + +func asStructuredError[K any](err error) (structuredError[K], bool) { + switch t := err.(type) { + case structuredError[K]: + return t, true + case *structuredError[K]: + if t == nil { + return structuredError[K]{}, false + } + + return *t, true + default: + return structuredError[K]{}, false + } +} + +func (i structuredError[K]) AsErrorExtension() ErrorExtension { + attrs := i.attributes.AsStringMap() + m := make(ErrorExtension, len(attrs)+5) + + for key, v := range attrs { + switch key { + // NOTE: skip reserved keys + case "field", "code", "component", "severity", "message": + // skip + default: + m[key] = v + } + } + + if path := i.Field().JSONPath(); path != "" { + m["field"] = i.field + } + + if i.code != "" { + m["code"] = i.code + } + + if i.component != "" { + m["component"] = string(i.component) + } + + m["message"] = i.message + m["severity"] = i.severity.String() + + return m +} + +type ErrorOption func(*Error) + +func WithAttribute(key any, value any) ErrorOption { + if key == nil { + panic("validation issue attribute key must not be nil") + } + + if t := reflect.TypeOf(key); t == nil || !t.Comparable() { + panic("validation issue attribute key is not comparable") + } + + return func(i *Error) { + i.attributes = i.attributes.Merge(Attributes{key: value}) + } +} + +func WithAttributes(attrs Attributes) ErrorOption { + return func(i *Error) { + if len(attrs) == 0 { + i.attributes = nil + return + } + + i.attributes = attrs + } +} + +func WithComponent(component ComponentName) ErrorOption { + return func(i *Error) { + i.component = component + } +} + +func WithField(parts ...*FieldDescriptor) ErrorOption { + return func(i *Error) { + sels := NewFieldSelectorGroup(parts...) + + i.field = sels + } +} + +func WithFieldString(parts ...string) ErrorOption { + return func(i *Error) { + sels := NewFieldSelectorGroup(lo.Map(parts, func(item string, _ int) *FieldDescriptor { + return NewFieldSelector(item) + })...) + + i.field = sels + } +} + +func WithSeverity(severity ErrorSeverity) ErrorOption { + return func(i *Error) { + i.severity = severity + } +} + +func WithCriticalSeverity() ErrorOption { + return WithSeverity(ErrorSeverityCritical) +} + +func WithWarningSeverity() ErrorOption { + return WithSeverity(ErrorSeverityWarning) +} + +func withMessage(message string) ErrorOption { + return func(i *Error) { + i.message = message + } +} + +func withCode(code ErrorCode) ErrorOption { + return func(i *Error) { + i.code = code + } +} + +func newStructuredError[K any](opts ...ErrorOption) structuredError[K] { + i := structuredError[K]{} + + for _, opt := range opts { + opt((*Error)(&i)) + } + + return i +} + +// NewError creates a structured system error. It is not a validation issue. +func NewError(code ErrorCode, message string, opts ...ErrorOption) Error { + return newStructuredError[generalErrorKind](append([]ErrorOption{withCode(code), withMessage(message)}, opts...)...) +} + +// WithCause retains the underlying error for errors.Is/As and errors.Unwrap. +func WithCause(cause error) ErrorOption { + return func(i *Error) { i.cause = cause } +} diff --git a/pkg/models/structured_error_test.go b/pkg/models/structured_error_test.go new file mode 100644 index 0000000000..80cd14207f --- /dev/null +++ b/pkg/models/structured_error_test.go @@ -0,0 +1,108 @@ +package models + +import ( + "encoding/json" + "errors" + "fmt" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestStructuredErrorFluentMethodsPreserveKind(t *testing.T) { + var ( + system Error + validation ValidationIssue + ) + + // These assignments also check the return type of every fluent method. + system = NewError("ledger_failed", "accounting failed"). + Clone().With(WithComponent("ledger")).WithField(NewFieldSelector("amount")). + WithPathString("source", "amount").WithSeverity(ErrorSeverityCritical). + WithAttr("reason", "missing_source").WithAttrs(Attributes{"amount": 10}). + SetAttributes(Attributes{"reason": "missing_source"}) + validation = NewValidationIssue("invalid_amount", "amount is invalid"). + Clone().With(WithComponent("ledger")).WithField(NewFieldSelector("amount")). + WithPathString("source", "amount").WithSeverity(ErrorSeverityWarning). + WithAttr("reason", "negative_amount").WithAttrs(Attributes{"amount": -10}). + SetAttributes(Attributes{"reason": "negative_amount"}) + + var issue ValidationIssue + require.False(t, errors.As(system, &issue)) + require.ErrorAs(t, fmt.Errorf("validate: %w", validation), &issue) + require.Equal(t, "negative_amount", issue.Attributes()["reason"]) + require.Equal(t, ErrorSeverityWarning, issue.Severity()) + + _, err := AsValidationIssues(system) + require.ErrorIs(t, err, system) + issues, err := AsValidationIssues(validation) + require.NoError(t, err) + require.Equal(t, "invalid_amount", string(issues[0].Code())) +} + +func TestStructuredErrorIdentityAndCause(t *testing.T) { + cause := errors.New("storage unavailable") + base := NewError("ledger_failed", "cannot load original entry", WithCause(cause)) + err := base.WithAttr("reason", "source_missing").With(WithComponent("ledger")) + + require.ErrorIs(t, err, base) + require.ErrorIs(t, fmt.Errorf("collect: %w", err), cause) + require.Same(t, cause, errors.Unwrap(err)) + require.Same(t, cause, err.Cause()) + require.Empty(t, base.Attributes()) + require.Equal(t, base.Error(), err.Error()) + + // Replacing a cause must not leave the old one reachable through fluent copies. + replacement := errors.New("another failure") + changed := err.With(WithCause(replacement)) + require.ErrorIs(t, changed, base) + require.ErrorIs(t, changed, replacement) + require.NotErrorIs(t, changed, cause) + require.NoError(t, errors.Unwrap(changed.With(WithCause(nil)))) + + validation := NewValidationIssue(base.Code(), base.Message()) + require.NotErrorIs(t, base, validation) + require.NotErrorIs(t, validation, base) +} + +func TestStructuredErrorDoesNotBecomeValidation(t *testing.T) { + issue := NewValidationIssue("invalid_amount", "amount is invalid") + system := NewError("ledger_failed", "accounting failed", WithCause(issue)).WithAttr("reason", "missing_source") + + for _, tc := range []struct { + name string + err error + }{ + {name: "value", err: system}, + {name: "pointer", err: &system}, + {name: "formatted wrapper", err: fmt.Errorf("collect: %w", system)}, + {name: "component wrapper", err: ErrorWithComponent("ledger", system)}, + {name: "field wrapper", err: ErrorWithFieldPrefix(NewFieldSelector("amount"), system)}, + {name: "joined with validation", err: errors.Join(issue, system)}, + } { + t.Run(tc.name, func(t *testing.T) { + issues, err := AsValidationIssues(tc.err) + require.ErrorIs(t, err, system) + require.Empty(t, issues) + }) + } + + // Error inspection can still reach the validation cause without classifying + // the enclosing system failure as a validation result. + var cause ValidationIssue + require.ErrorAs(t, system, &cause) + require.Equal(t, issue, cause) +} + +func TestStructuredErrorJSONAndAttributeIsolation(t *testing.T) { + err := NewError("ledger_failed", "accounting failed", WithComponent("ledger"), WithCause(errors.New("private cause"))). + WithAttr("reason", "missing_source") + attrs := err.Attributes() + attrs["reason"] = "changed" + require.Equal(t, "missing_source", err.Attributes()["reason"]) + + encoded, marshalErr := json.Marshal(err) + require.NoError(t, marshalErr) + require.JSONEq(t, `{"code":"ledger_failed","message":"accounting failed","component":"ledger","severity":"critical","reason":"missing_source"}`, string(encoded)) + require.NotContains(t, string(encoded), "private cause") +} diff --git a/pkg/models/testutils.go b/pkg/models/testutils.go index aeeb7ca273..a1b6c86a49 100644 --- a/pkg/models/testutils.go +++ b/pkg/models/testutils.go @@ -25,8 +25,8 @@ func RequireValidationIssuesMatch(t *testing.T, expected ValidationIssues, actua actCopy[i].field = nil expCopy[i].field = nil - actCopy[i].wraps = nil - expCopy[i].wraps = nil + actCopy[i].previous = nil + expCopy[i].previous = nil } require.Equalf(t, expCopy, actCopy, "issues must match") diff --git a/pkg/models/validationissue.go b/pkg/models/validationissue.go index 29d0d3fe22..c395e362f7 100644 --- a/pkg/models/validationissue.go +++ b/pkg/models/validationissue.go @@ -3,342 +3,31 @@ package models import ( "encoding/json" "errors" - "maps" - "reflect" "github.com/samber/lo" ) -type ErrorExtension map[string]any - var ( - _ error = (*ValidationIssue)(nil) - _ json.Marshaler = (*ValidationIssue)(nil) - _ Equaler[ValidationIssue] = (*ValidationIssue)(nil) + _ error = ValidationIssue{} + _ json.Marshaler = ValidationIssue{} + _ Equaler[ValidationIssue] = ValidationIssue{} ) -type ValidationIssue struct { - attributes Attributes - code ErrorCode - component ComponentName - message string - field *FieldDescriptor - severity ErrorSeverity - wraps error -} - -func (i ValidationIssue) MarshalJSON() ([]byte, error) { - m := i.AsErrorExtension() - - return json.Marshal(m) -} - -func (i ValidationIssue) Clone() ValidationIssue { - return ValidationIssue{ - attributes: i.attributes.Clone(), - code: i.code, - component: i.component, - message: i.message, - field: i.field, - severity: i.severity, - wraps: i, - } -} - -// Attributes returns a shallow-copy of the underlying attributes map. -func (i ValidationIssue) Attributes() Attributes { - return maps.Clone(i.attributes) -} - -func (i ValidationIssue) Code() ErrorCode { - return i.code -} - -func (i ValidationIssue) Component() ComponentName { - return i.component -} - -func (i ValidationIssue) Message() string { - return i.message -} - -func (i ValidationIssue) Field() *FieldDescriptor { - if i.field == nil { - return NewFieldSelectorGroup() - } - - return i.field -} - -func (i ValidationIssue) Severity() ErrorSeverity { - return i.severity -} - -func (i ValidationIssue) With(opts ...ValidationIssueOption) ValidationIssue { - v := i.Clone() - for _, opt := range opts { - opt(&v) - } - - return v -} - -func (i ValidationIssue) WithField(parts ...*FieldDescriptor) ValidationIssue { - v := i.Clone() - group := NewFieldSelectorGroup(parts...) - - v.field = group - - return v -} - -func (i ValidationIssue) WithPathString(parts ...string) ValidationIssue { - return i.WithField(lo.Map(parts, func(item string, _ int) *FieldDescriptor { - return NewFieldSelector(item) - })...) -} - -func (i ValidationIssue) WithSeverity(s ErrorSeverity) ValidationIssue { - v := i.Clone() - v.severity = s - - return v -} - -func (i ValidationIssue) WithAttr(key any, value any) ValidationIssue { - if key == nil { - panic("validation issue attribute key must not be nil") - } - - if t := reflect.TypeOf(key); t == nil || !t.Comparable() { - panic("validation issue attribute key is not comparable") - } - - return i.WithAttrs(Attributes{ - key: value, - }) -} - -func (i ValidationIssue) WithAttrs(attrs Attributes) ValidationIssue { - if len(attrs) == 0 { - return i - } - - for k := range attrs { - if k == nil { - panic("validation issue attribute key must not be nil") - } - - if t := reflect.TypeOf(k); t == nil || !t.Comparable() { - panic("validation issue attribute key is not comparable") - } - } - - v := i.Clone() - v.attributes = i.attributes.Merge(attrs) - - return v -} +// ValidationIssue is a structured error that callers may handle as domain validation. +type ValidationIssue = structuredError[validationIssueKind] -func (i ValidationIssue) SetAttributes(attrs Attributes) ValidationIssue { - v := i.Clone() +type validationIssueKind struct{} - v.attributes = attrs +// ValidationIssueOption keeps existing option declarations compatible. +type ValidationIssueOption = ErrorOption - return v -} - -func (i ValidationIssue) Error() string { - return i.message -} - -// Is is required in addition to Unwrap because ValidationIssue is not comparable: -// it contains map-backed attributes, so errors.Is cannot use == when it reaches a -// wrapped ValidationIssue in the chain. -func (i ValidationIssue) Is(target error) bool { - targetIssue, ok := asValidationIssue(target) - if !ok { - return false - } - - return i.Equal(targetIssue) -} - -func (i ValidationIssue) Unwrap() error { - return i.wraps -} - -func (i ValidationIssue) Equal(other ValidationIssue) bool { - return maps.EqualFunc(i.attributes, other.attributes, equalValidationIssueAttributeValue) && - i.code == other.code && - i.component == other.component && - i.message == other.message && - i.severity == other.severity && - i.Field().String() == other.Field().String() -} - -// equalValidationIssueAttributeValue is used to compare attribute values for equality. -// It's better than reflect.DeepEqual because it handles nil values properly (DeepEqual would consider nil and empty value equal) -func equalValidationIssueAttributeValue(left, right any) bool { - if left == nil || right == nil { - return left == right - } - - // Let's not use DeepEqual for comparable types - if t := reflect.TypeOf(left); t != nil && t.Comparable() { - return left == right - } - - return reflect.DeepEqual(left, right) -} - -func asValidationIssue(err error) (ValidationIssue, bool) { - switch t := err.(type) { - case ValidationIssue: - return t, true - case *ValidationIssue: - if t == nil { - return ValidationIssue{}, false - } - - return *t, true - default: - return ValidationIssue{}, false - } -} - -func (i ValidationIssue) AsErrorExtension() ErrorExtension { - attrs := i.attributes.AsStringMap() - m := make(ErrorExtension, len(attrs)+5) - - for key, v := range attrs { - switch key { - // NOTE: skip reserved keys - case "field", "code", "component", "severity", "message": - // skip - default: - m[key] = v - } - } - - if path := i.Field().JSONPath(); path != "" { - m["field"] = i.field - } - - if i.code != "" { - m["code"] = i.code - } - - if i.component != "" { - m["component"] = string(i.component) - } - - m["message"] = i.message - m["severity"] = i.severity.String() - - return m -} - -type ValidationIssueOption func(*ValidationIssue) - -func WithAttribute(key any, value any) ValidationIssueOption { - if key == nil { - panic("validation issue attribute key must not be nil") - } - - if t := reflect.TypeOf(key); t == nil || !t.Comparable() { - panic("validation issue attribute key is not comparable") - } - - return func(i *ValidationIssue) { - i.attributes = i.attributes.Merge(Attributes{key: value}) - } -} - -func WithAttributes(attrs Attributes) ValidationIssueOption { - return func(i *ValidationIssue) { - if len(attrs) == 0 { - i.attributes = nil - return - } - - i.attributes = attrs - } -} - -func WithComponent(component ComponentName) ValidationIssueOption { - return func(i *ValidationIssue) { - i.component = component - } -} - -func WithField(parts ...*FieldDescriptor) ValidationIssueOption { - return func(i *ValidationIssue) { - sels := NewFieldSelectorGroup(parts...) - - i.field = sels - } -} - -func WithFieldString(parts ...string) ValidationIssueOption { - return func(i *ValidationIssue) { - sels := NewFieldSelectorGroup(lo.Map(parts, func(item string, _ int) *FieldDescriptor { - return NewFieldSelector(item) - })...) - - i.field = sels - } -} - -func WithSeverity(severity ErrorSeverity) ValidationIssueOption { - return func(i *ValidationIssue) { - i.severity = severity - } -} - -func WithCriticalSeverity() ValidationIssueOption { - return WithSeverity(ErrorSeverityCritical) -} - -func WithWarningSeverity() ValidationIssueOption { - return WithSeverity(ErrorSeverityWarning) -} - -func withMessage(message string) ValidationIssueOption { - return func(i *ValidationIssue) { - i.message = message - } -} - -func withCode(code ErrorCode) ValidationIssueOption { - return func(i *ValidationIssue) { - i.code = code - } +// NewValidationIssue creates a structured domain validation issue. +func NewValidationIssue(code ErrorCode, message string, opts ...ValidationIssueOption) ValidationIssue { + return newValidationIssue(append([]ErrorOption{withCode(code), withMessage(message)}, opts...)...) } -// We internally construct each ValidationIssue with this private constructor -// so Option Functions are always used to guarantee correct creation of the ValidationIssue. func newValidationIssue(opts ...ValidationIssueOption) ValidationIssue { - i := ValidationIssue{} - - for _, opt := range opts { - opt(&i) - } - - return i -} - -// NewValidationIssue returns a new ValidationIssue with code and message. -func NewValidationIssue(code ErrorCode, message string, opts ...ValidationIssueOption) ValidationIssue { - return newValidationIssue( - append( - []ValidationIssueOption{ - withMessage(message), - withCode(code), - }, - opts..., - )..., - ) + return newStructuredError[validationIssueKind](opts...) } // NewValidationError returns a new ValidationIssue with code and message and its severity set to SeverityCritical. @@ -384,9 +73,9 @@ func (v ValidationIssues) WithSeverityOrHigher(severity ErrorSeverity) Validatio }) } -// AsValidationIssues returns a list of ValidationIssue from the input error or the errIn error in case: -// * the errIn is `nil` -// * any leaf errors in the error tree that are not wrapped with WrapWithComponent or WrapWithFieldPrefix functions are treated as critical errors +// AsValidationIssues extracts issues through wrappers and joins. If any branch +// contains a system error, it returns the original error tree. Legacy field and +// component wrappers still convert plain errors; a structured Error stays a failure. func AsValidationIssues(errIn error) (ValidationIssues, error) { if errIn == nil { return nil, nil @@ -410,6 +99,15 @@ func asValidationIssues(err error, prefix *FieldDescriptor, component ComponentN return asValidationIssues(e.err, prefix, e.component, true) case fieldPrefixedWrapper: return asValidationIssues(e.err, e.prefix.WithPrefix(prefix), component, true) + case Error: + // A structured system error is a failure even if its cause is a validation issue. + return nil, err + case *Error: + if e == nil { + return nil, nil + } + + return nil, err case ValidationIssue: return validationIssueAsValidationIssues(e, prefix, component), nil case *ValidationIssue: diff --git a/pkg/models/validationissue_test.go b/pkg/models/validationissue_test.go index c491543db3..0fafa3612a 100644 --- a/pkg/models/validationissue_test.go +++ b/pkg/models/validationissue_test.go @@ -156,13 +156,13 @@ func TestValidationIssue_Clone(t *testing.T) { message: errTestValidationIssue.message, field: errTestValidationIssue.field, severity: errTestValidationIssue.severity, - wraps: errTestValidationIssue, + previous: errTestValidationIssue, }, }, { name: "empty", issue: ValidationIssue{}, - expectedIssue: ValidationIssue{wraps: ValidationIssue{}}, + expectedIssue: ValidationIssue{previous: ValidationIssue{}}, }, } diff --git a/test/app/stripe/invoice_test.go b/test/app/stripe/invoice_test.go index b6dd163825..18e80dc355 100644 --- a/test/app/stripe/invoice_test.go +++ b/test/app/stripe/invoice_test.go @@ -15,6 +15,7 @@ import ( "github.com/stretchr/testify/mock" "github.com/stretchr/testify/suite" "github.com/stripe/stripe-go/v80" + metricnoop "go.opentelemetry.io/otel/metric/noop" "github.com/openmeterio/openmeter/openmeter/app" appstripe "github.com/openmeterio/openmeter/openmeter/app/stripe" @@ -135,6 +136,7 @@ func (s *StripeInvoiceTestSuite) SetupSuite() { advanceService := advancetestutils.NewService(s.T(), ledgerDeps, breakageService) collectorService, err := ledgercollector.NewService(ledgercollector.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: omtestutils.NewDiscardLogger(s.T()), Advance: advanceService, Ledger: ledgerDeps.HistoricalLedger, diff --git a/test/billing/lineengine_test.go b/test/billing/lineengine_test.go index 55e30e9a79..f20346fad8 100644 --- a/test/billing/lineengine_test.go +++ b/test/billing/lineengine_test.go @@ -700,62 +700,78 @@ func (s *LineEngineTestSuite) TestGatheringPreviewUsesPreviewLineEngineCallback( } func (s *LineEngineTestSuite) TestCollectionCompletedSystemErrorsAbortCollection() { - var ( - ctx = s.T().Context() - namespace = s.GetUniqueNamespace("ns-line-engine-collection-completed-validation") - mockEngine = &mockCollectionCompletedLineEngine{engineType: ombilling.LineEngineTypeChargeUsageBased} - invoice ombilling.StandardInvoice - collectionAt time.Time - err error - ) + for _, tc := range []struct { + name string + err error + }{ + {name: "plain error", err: errors.New("mock collection completed failure")}, + { + name: "structured system error with validation cause", + err: fmt.Errorf("collect: %w", models.NewError("ledger_failed", "mock collection completed failure", + models.WithCause(models.NewValidationIssue("invalid_source", "source is invalid")), + ).WithAttr("reason", "source_missing")), + }, + } { + s.Run(tc.name, func() { + var ( + ctx = s.T().Context() + namespace = s.GetUniqueNamespace("ns-line-engine-collection-completed-validation") + mockEngine = &mockCollectionCompletedLineEngine{engineType: ombilling.LineEngineTypeChargeUsageBased} + invoice ombilling.StandardInvoice + collectionAt time.Time + err error + ) - clockBase := lo.Must(time.Parse(time.RFC3339, "2024-09-02T12:13:14Z")) - clock.SetTime(clockBase) - defer clock.ResetTime() - defer func() { _ = s.MeterAdapter.ReplaceMeters(ctx, []meter.Meter{}) }() - defer s.MockStreamingConnector.Reset() - s.registerMockLineEngine(s.T(), mockEngine) - defer s.unregisterLineEngine(s.T(), mockEngine) + clockBase := lo.Must(time.Parse(time.RFC3339, "2024-09-02T12:13:14Z")) + clock.SetTime(clockBase) + defer clock.ResetTime() + defer func() { _ = s.MeterAdapter.ReplaceMeters(ctx, []meter.Meter{}) }() + defer s.MockStreamingConnector.Reset() + s.registerMockLineEngine(s.T(), mockEngine) + defer s.unregisterLineEngine(s.T(), mockEngine) - s.Run("Given a draft invoice waiting for collection with a failing collection-completed engine", func() { - defer mockEngine.Reset() + s.Run("Given a draft invoice waiting for collection with a failing collection-completed engine", func() { + defer mockEngine.Reset() - mockEngine.buildStandardInvoiceLines = func(_ context.Context, input ombilling.BuildStandardInvoiceLinesInput) (ombilling.StandardLines, error) { - return mustAsNewStandardLines(ombilling.BuildStandardInvoiceLinesInput{ - Invoice: input.Invoice, - GatheringLines: input.GatheringLines, - }), nil - } + mockEngine.buildStandardInvoiceLines = func(_ context.Context, input ombilling.BuildStandardInvoiceLinesInput) (ombilling.StandardLines, error) { + return mustAsNewStandardLines(ombilling.BuildStandardInvoiceLinesInput{ + Invoice: input.Invoice, + GatheringLines: input.GatheringLines, + }), nil + } - invoice, collectionAt = s.createMeteredDraftInvoiceWaitingForCollection( - ctx, - namespace, - mockEngine.GetLineEngineType(), - "UBP - mock collection completed", - ) - }) + invoice, collectionAt = s.createMeteredDraftInvoiceWaitingForCollection( + ctx, + namespace, + mockEngine.GetLineEngineType(), + "UBP - mock collection completed", + ) + }) - s.Run("When collection is completed", func() { - defer mockEngine.Reset() + s.Run("When collection is completed", func() { + defer mockEngine.Reset() - mockEngine.onCollectionCompleted = func(_ context.Context, input ombilling.OnCollectionCompletedInput) (ombilling.StandardLines, error) { - return nil, fmt.Errorf("mock collection completed failure") - } + mockEngine.onCollectionCompleted = func(_ context.Context, input ombilling.OnCollectionCompletedInput) (ombilling.StandardLines, error) { + return nil, tc.err + } - clock.SetTime(collectionAt.Add(time.Minute)) - _, err = s.BillingService.AdvanceInvoice(ctx, invoice.GetInvoiceID()) - s.Require().ErrorContains(err, "mock collection completed failure") - }) + clock.SetTime(collectionAt.Add(time.Minute)) + _, err = s.BillingService.AdvanceInvoice(ctx, invoice.GetInvoiceID()) + s.Require().ErrorContains(err, "mock collection completed failure") + s.Require().ErrorIs(err, tc.err) + }) - s.Run("Then collection is rolled back without recording a validation issue", func() { - invoice, err = s.BillingService.GetStandardInvoiceById(ctx, ombilling.GetStandardInvoiceByIdInput{ - Invoice: invoice.GetInvoiceID(), + s.Run("Then collection is rolled back without recording a validation issue", func() { + invoice, err = s.BillingService.GetStandardInvoiceById(ctx, ombilling.GetStandardInvoiceByIdInput{ + Invoice: invoice.GetInvoiceID(), + }) + s.Require().NoError(err) + s.Equal(ombilling.StandardInvoiceStatusDraftWaitingForCollection, invoice.Status) + s.Empty(invoice.ValidationIssues) + s.Nil(invoice.QuantitySnapshotedAt) + }) }) - s.Require().NoError(err) - s.Equal(ombilling.StandardInvoiceStatusDraftWaitingForCollection, invoice.Status) - s.Empty(invoice.ValidationIssues) - s.Nil(invoice.QuantitySnapshotedAt) - }) + } } func (s *LineEngineTestSuite) TestCollectionCompletedValidationIssuesRespectSeverity() { diff --git a/test/credits/base.go b/test/credits/base.go index 7ee556af44..290ae485e0 100644 --- a/test/credits/base.go +++ b/test/credits/base.go @@ -10,6 +10,7 @@ import ( "github.com/invopop/gobl/currency" "github.com/samber/lo" "github.com/samber/mo" + metricnoop "go.opentelemetry.io/otel/metric/noop" "github.com/openmeterio/openmeter/openmeter/billing" "github.com/openmeterio/openmeter/openmeter/billing/charges" @@ -149,6 +150,7 @@ func (s *BaseSuite) SetupSuite() { s.RevenueRecognizer = revenueRecognizer collectorService, err := ledgercollector.NewService(ledgercollector.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: logger, Advance: s.AdvanceService, Ledger: deps.HistoricalLedger, diff --git a/test/credits_billing_compat/suite.go b/test/credits_billing_compat/suite.go index 01c0cf2f6c..e049b2480a 100644 --- a/test/credits_billing_compat/suite.go +++ b/test/credits_billing_compat/suite.go @@ -1,6 +1,8 @@ package creditsbillingcompat import ( + metricnoop "go.opentelemetry.io/otel/metric/noop" + "github.com/openmeterio/openmeter/openmeter/billing" "github.com/openmeterio/openmeter/openmeter/billing/charges" "github.com/openmeterio/openmeter/openmeter/billing/charges/creditpurchase" @@ -171,6 +173,7 @@ func (s *BaseSuite) setupCreditsSuite() { s.RevenueRecognizer = revenueRecognizer collectorService, err := ledgercollector.NewService(ledgercollector.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: logger, Advance: s.AdvanceService, Ledger: deps.HistoricalLedger, diff --git a/test/subscription/framework_test.go b/test/subscription/framework_test.go index 3a0adf5976..72ac42df7c 100644 --- a/test/subscription/framework_test.go +++ b/test/subscription/framework_test.go @@ -210,6 +210,7 @@ func setup(t *testing.T, config setupConfig) testDeps { advanceService := advancetestutils.NewService(t, ledgerDeps, breakageService) collector, err := ledgercollector.NewService(ledgercollector.Config{ + Meter: metricnoop.NewMeterProvider().Meter("test"), Logger: logger, Advance: advanceService, Ledger: ledgerDeps.HistoricalLedger,