From c6ec6eb4fa3a108de108b5f18bb512e5706bdcdd Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Fri, 10 Jul 2026 17:57:56 +0200 Subject: [PATCH 01/13] perf(customer): rewrite bulk usage attribution lookup as UNION ALL GetCustomersByUsageAttribution had the same cross-table OR anti-pattern PR #4684 fixed for the single-key lookup, causing a Postgres seq scan on large namespaces. Split into two independently-indexable UNION ALL branches, same as the single-key fix, without lookup_priority (bulk still returns every match). Benchmark: 222ms -> 1ms (220x) at 100 keys / 100k customers. --- openmeter/customer/adapter/customer.go | 82 ++++++++++--- ...stomer_usage_attribution_benchmark_test.go | 110 ++++++++++++++++++ 2 files changed, 176 insertions(+), 16 deletions(-) diff --git a/openmeter/customer/adapter/customer.go b/openmeter/customer/adapter/customer.go index 865eb6416b..7334a01752 100644 --- a/openmeter/customer/adapter/customer.go +++ b/openmeter/customer/adapter/customer.go @@ -613,6 +613,68 @@ func customerMatchesUsageAttributionKey(namespace, key string, at time.Time) pre } } +// customersMatchUsageAttributionKeys is the bulk counterpart of customerMatchesUsageAttributionKey. +// It resolves customer IDs whose own key OR one of their subject keys is in the given key set, +// via the same two independently-indexable UNION ALL branches, but returns the full candidate set +// rather than picking one winner per key: a key that matches both a customer's own key and a +// different customer's subject key legitimately resolves to two distinct customers here, and it is +// the caller's responsibility to apply key-over-subject precedence per input key (see +// customer/service.GetCustomersByUsageAttribution). The generated SQL shape mirrors: +// +// WHERE customers.id IN ( +// SELECT c.id FROM customers AS c +// WHERE c.namespace = $1 AND c.key IN (...) AND c.deleted_at IS NULL +// +// UNION ALL +// +// SELECT c.id +// FROM customer_subjects AS cs +// JOIN customers AS c ON c.id = cs.customer_id +// WHERE cs.namespace = $1 AND cs.subject_key IN (...) +// AND cs.created_at <= $2 +// AND (cs.deleted_at IS NULL OR cs.deleted_at > $2) +// AND c.namespace = $1 AND c.deleted_at IS NULL +// ) +func customersMatchUsageAttributionKeys(namespace string, keys []string, at time.Time) predicate.Customer { + return func(s *sql.Selector) { + keyValues := lo.ToAnySlice(keys) + + keyCustomerTable := sql.Table(customerdb.Table).As("customer_by_key") + customerKeyMatch := sql.Select(keyCustomerTable.C(customerdb.FieldID)). + From(keyCustomerTable). + Where(sql.And( + sql.EQ(keyCustomerTable.C(customerdb.FieldNamespace), namespace), + sql.In(keyCustomerTable.C(customerdb.FieldKey), keyValues...), + sql.IsNull(keyCustomerTable.C(customerdb.FieldDeletedAt)), + )) + + customerSubjectsTable := sql.Table(customersubjectsdb.Table).As("customer_subjects") + subjectCustomerTable := sql.Table(customerdb.Table).As("customer_by_subject") + subjectKeyMatch := sql.Select(subjectCustomerTable.C(customerdb.FieldID)). + From(customerSubjectsTable). + Join(subjectCustomerTable). + On( + subjectCustomerTable.C(customerdb.FieldID), + customerSubjectsTable.C(customersubjectsdb.FieldCustomerID), + ). + Where(sql.And( + sql.EQ(customerSubjectsTable.C(customersubjectsdb.FieldNamespace), namespace), + sql.In(customerSubjectsTable.C(customersubjectsdb.FieldSubjectKey), keyValues...), + sql.LTE(customerSubjectsTable.C(customersubjectsdb.FieldCreatedAt), at), + sql.Or( + sql.IsNull(customerSubjectsTable.C(customersubjectsdb.FieldDeletedAt)), + sql.GT(customerSubjectsTable.C(customersubjectsdb.FieldDeletedAt), at), + ), + sql.EQ(subjectCustomerTable.C(customerdb.FieldNamespace), namespace), + sql.IsNull(subjectCustomerTable.C(customerdb.FieldDeletedAt)), + )) + + candidates := customerKeyMatch.UnionAll(subjectKeyMatch) + + s.Where(sql.In(s.C(customerdb.FieldID), candidates)) + } +} + // GetCustomersByUsageAttribution resolves multiple customers by usage attribution keys in a single query. // A key matches a customer either by the customer's own key or by one of its subject keys, mirroring // the single-key GetCustomerByUsageAttribution. Keys that match no customer are simply absent from the @@ -624,23 +686,11 @@ func (a *adapter) GetCustomersByUsageAttribution(ctx context.Context, input cust // TODO: consider adding a CreatedAtLTE(now) guard to the customer/subject query, // to defend against the edge case of having customers/subjects becoming active in the future query := repo.db.Customer.Query(). - Where(customerdb.Namespace(input.Namespace)). Where( - customerdb.Or( - // We lookup customers by subject key in the subjects table - customerdb.HasSubjectsWith( - customersubjectsdb.SubjectKeyIn(input.Keys...), - customersubjectsdb.CreatedAtLTE(now), - customersubjectsdb.Or( - customersubjectsdb.DeletedAtIsNil(), - customersubjectsdb.DeletedAtGT(now), - ), - ), - // Or else we lookup customers by key in the customers table - customerdb.KeyIn(input.Keys...), - ), - ). - Where(customerdb.DeletedAtIsNil()) + customerdb.Namespace(input.Namespace), + customerdb.DeletedAtIsNil(), + customersMatchUsageAttributionKeys(input.Namespace, input.Keys, now), + ) // TODO: consider adding a CreatedAtLTE(now) guard to the WithSubjects query, // to defend against the edge case of having customers/subjects becoming active in the future diff --git a/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go b/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go index aeaa8f7d9a..8d97c3a329 100644 --- a/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go +++ b/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go @@ -122,6 +122,116 @@ func BenchmarkCustomerUsageAttributionLookup(b *testing.B) { } } +// bulkUsageAttributionBenchmarkKeyCount mirrors the governance QueryAccess OAS cap +// (@maxItems(100) on customer.keys), the primary bulk consumer's realistic upper bound. +const bulkUsageAttributionBenchmarkKeyCount = 100 + +// BenchmarkCustomersUsageAttributionBulkLookup compares the cross-table OR with the UNION ALL +// candidate-ID lookup for the bulk (multi-key) resolution path, mirroring +// BenchmarkCustomerUsageAttributionLookup for the single-key path. Seeding is outside the timed +// sections. Set CUSTOMER_USAGE_ATTRIBUTION_BENCH_CUSTOMERS to change the namespace size. +func BenchmarkCustomersUsageAttributionBulkLookup(b *testing.B) { + b.StopTimer() + + db := testutils.InitPostgresDB(b) + b.Cleanup(func() { db.Close(b) }) + + client := db.EntDriver.Client() + if err := client.Schema.Create(b.Context()); err != nil { + b.Fatalf("create database schema: %v", err) + } + + customerCount := usageAttributionBenchmarkCustomerCount(b) + if customerCount < 2*bulkUsageAttributionBenchmarkKeyCount { + b.Fatalf("%s must be at least %d to seed a non-overlapping bulk key set, got %d", + usageAttributionBenchmarkSizeEnv, 2*bulkUsageAttributionBenchmarkKeyCount, customerCount) + } + + namespace := "customer-usage-attribution-bulk-benchmark" + seedUsageAttributionBenchmark(b, client, namespace, customerCount) + + now := time.Now().UTC() + + // Half the keys resolve via a direct customer-key match, half via a subject-key match on a + // disjoint customer range, mirroring a realistic mixed bulk lookup (governance resolves + // customer keys and subject keys through the same call). + half := bulkUsageAttributionBenchmarkKeyCount / 2 + keys := make([]string, 0, bulkUsageAttributionBenchmarkKeyCount) + for i := 1; i <= half; i++ { + keys = append(keys, fmt.Sprintf("customer-%d", i)) + } + for i := half + 1; i <= 2*half; i++ { + keys = append(keys, fmt.Sprintf("subject-%d", i)) + } + + variants := []struct { + name string + query func(context.Context) (int, error) + }{ + { + name: "cross_table_or", + query: func(ctx context.Context) (int, error) { + return client.Customer.Query(). + Where( + customerdb.Namespace(namespace), + customerdb.Or( + customerdb.HasSubjectsWith( + customersubjectsdb.SubjectKeyIn(keys...), + customersubjectsdb.Or( + customersubjectsdb.DeletedAtIsNil(), + customersubjectsdb.DeletedAtGT(now), + ), + ), + customerdb.KeyIn(keys...), + ), + customerdb.DeletedAtIsNil(), + ). + Count(ctx) + }, + }, + { + name: "union_all", + query: func(ctx context.Context) (int, error) { + return client.Customer.Query(). + Where( + customerdb.Namespace(namespace), + customerdb.DeletedAtIsNil(), + customersMatchUsageAttributionKeys(namespace, keys, now), + ). + Count(ctx) + }, + }, + } + + for _, variant := range variants { + b.Run(variant.name, func(b *testing.B) { + count, err := variant.query(b.Context()) + if err != nil { + b.Fatalf("warm up lookup: %v", err) + } + if count != bulkUsageAttributionBenchmarkKeyCount { + b.Fatalf("warm up lookup returned %d customers, expected %d", count, bulkUsageAttributionBenchmarkKeyCount) + } + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + count, err = variant.query(b.Context()) + if err != nil { + b.Fatalf("lookup customers: %v", err) + } + } + b.StopTimer() + b.ReportMetric(float64(customerCount), "customers") + b.ReportMetric(float64(bulkUsageAttributionBenchmarkKeyCount), "keys") + + if count != bulkUsageAttributionBenchmarkKeyCount { + b.Fatalf("lookup returned %d customers, expected %d", count, bulkUsageAttributionBenchmarkKeyCount) + } + }) + } +} + func usageAttributionBenchmarkCustomerCount(b *testing.B) int { b.Helper() From 5f490d3bb603b5c54359297ecf65cee4baa6add2 Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Fri, 10 Jul 2026 18:03:09 +0200 Subject: [PATCH 02/13] fix(customer): resolve usage-attribution key precedence deterministically MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GetCustomersByUsageAttribution returned an unordered []Customer, so callers (governance + a private-repo consumer) each built their own map[key]Customer with first-match-wins semantics that depended on DB row order — unfixable by reordering when two customers collide on the same key from opposite sides (one's own key vs. the other's subject key). Service now returns map[string]Customer with key-over-subject precedence resolved once, centrally, mirroring the single-key lookup's UNION ALL priority. Each ambiguous collision is logged with the key and both customer IDs for operator follow-up, replacing the duplicated (and now provably buggy) mapping loops in every consumer. --- app/common/customer.go | 1 + openmeter/customer/service.go | 2 +- openmeter/customer/service/customer.go | 89 +++++++++++++- openmeter/customer/service/customer_test.go | 116 ++++++++++++++++++ .../service/hooks/subjectcustomer_test.go | 1 + openmeter/customer/service/service.go | 8 ++ openmeter/customer/testutils/env.go | 1 + .../entitlement/metered/lateevents_test.go | 1 + openmeter/entitlement/metered/utils_test.go | 1 + openmeter/entitlement/service/utils_test.go | 1 + openmeter/governance/service/service.go | 30 +---- openmeter/governance/service/service_test.go | 1 + openmeter/server/server_test.go | 2 +- .../service/hooks/customersubject_test.go | 1 + openmeter/subject/testutils/env.go | 1 + openmeter/subscription/testutils/customer.go | 1 + test/app/stripe/testenv.go | 1 + test/app/testenv.go | 1 + test/billing/suite.go | 1 + test/customer/testenv.go | 1 + test/entitlement/regression/framework_test.go | 1 + 21 files changed, 232 insertions(+), 30 deletions(-) create mode 100644 openmeter/customer/service/customer_test.go diff --git a/app/common/customer.go b/app/common/customer.go index 9f465e6caa..b92bb80025 100644 --- a/app/common/customer.go +++ b/app/common/customer.go @@ -40,6 +40,7 @@ func NewCustomerService( service, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: eventPublisher, + Logger: logger.WithGroup("customer"), }) if err != nil { return nil, fmt.Errorf("failed to create customer service: %w", err) diff --git a/openmeter/customer/service.go b/openmeter/customer/service.go index f39fad415d..6d2aede5d0 100644 --- a/openmeter/customer/service.go +++ b/openmeter/customer/service.go @@ -26,6 +26,6 @@ type CustomerService interface { DeleteCustomer(ctx context.Context, customer DeleteCustomerInput) error GetCustomer(ctx context.Context, customer GetCustomerInput) (*Customer, error) GetCustomerByUsageAttribution(ctx context.Context, input GetCustomerByUsageAttributionInput) (*Customer, error) - GetCustomersByUsageAttribution(ctx context.Context, input GetCustomersByUsageAttributionInput) ([]Customer, error) + GetCustomersByUsageAttribution(ctx context.Context, input GetCustomersByUsageAttributionInput) (map[string]Customer, error) UpdateCustomer(ctx context.Context, params UpdateCustomerInput) (*Customer, error) } diff --git a/openmeter/customer/service/customer.go b/openmeter/customer/service/customer.go index 9c0d1ebf4c..1062ae5297 100644 --- a/openmeter/customer/service/customer.go +++ b/openmeter/customer/service/customer.go @@ -136,15 +136,98 @@ func (s *Service) GetCustomerByUsageAttribution(ctx context.Context, input custo return s.adapter.GetCustomerByUsageAttribution(ctx, input) } -// GetCustomersByUsageAttribution resolves multiple customers by usage attribution keys in a single query -func (s *Service) GetCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) ([]customer.Customer, error) { +// GetCustomersByUsageAttribution resolves multiple customers by usage attribution keys in a single +// query, mapping each input key to the customer it matches with key-over-subject precedence applied. +func (s *Service) GetCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) (map[string]customer.Customer, error) { if err := input.Validate(); err != nil { return nil, models.NewGenericValidationError( fmt.Errorf("error getting customers by usage attribution: %w", err), ) } - return s.adapter.GetCustomersByUsageAttribution(ctx, input) + customers, err := s.adapter.GetCustomersByUsageAttribution(ctx, input) + if err != nil { + return nil, err + } + + resolved, ambiguous := resolveCustomersByKey(customers, input.Keys) + + // A key resolving to both a distinct key-owner and a distinct subject-owner is not an error — + // the key-owner wins deterministically, mirroring GetCustomerByUsageAttribution's single-row + // precedence — but the underlying data (one key doing double duty as another customer's own + // key and a different customer's subject key) is unusual enough to warrant operator attention, + // so each occurrence is logged with the key and both customer IDs for follow-up. + for _, m := range ambiguous { + s.logger.ErrorContext(ctx, "ambiguous usage attribution key: matches both a customer key and a distinct subject key", + "namespace", input.Namespace, + "key", m.Key, + "key_owner.id", m.KeyOwnerID, + "subject_owner.id", m.SubjectOwnerID, + ) + } + + return resolved, nil +} + +// ambiguousUsageAttributionMatch records a key that matched both a distinct customer-key owner and +// a distinct subject-key owner, so the caller can log the identifiers needed to investigate and fix +// the underlying data overlap. +type ambiguousUsageAttributionMatch struct { + Key string + KeyOwnerID string + SubjectOwnerID string +} + +// resolveCustomersByKey maps each input key to the customer it matches. A key matches a customer +// either by the customer's own key or by one of its subject keys; when a key matches both a +// distinct key-owner and a distinct subject-owner, the key-owner takes precedence, mirroring the +// single-key GetCustomerByUsageAttribution's UNION ALL lookup_priority ordering. Keys with no match +// are absent from the returned map. The second return value lists the keys where precedence +// actually mattered (a distinct key-owner and a distinct subject-owner both matched). +func resolveCustomersByKey(customers []customer.Customer, keys []string) (map[string]customer.Customer, []ambiguousUsageAttributionMatch) { + byKey := make(map[string]customer.Customer, len(customers)) + bySubject := make(map[string]customer.Customer, len(customers)) + + for _, c := range customers { + if c.Key != nil { + byKey[*c.Key] = c + } + + if c.UsageAttribution != nil { + for _, sk := range c.UsageAttribution.SubjectKeys { + if _, ok := bySubject[sk]; !ok { + bySubject[sk] = c + } + } + } + } + + resolved := make(map[string]customer.Customer, len(keys)) + var ambiguous []ambiguousUsageAttributionMatch + + for _, k := range keys { + keyOwner, hasKeyOwner := byKey[k] + subjectOwner, hasSubjectOwner := bySubject[k] + + if hasKeyOwner && hasSubjectOwner && keyOwner.ID != subjectOwner.ID { + ambiguous = append(ambiguous, ambiguousUsageAttributionMatch{ + Key: k, + KeyOwnerID: keyOwner.ID, + SubjectOwnerID: subjectOwner.ID, + }) + } + + if hasKeyOwner { + resolved[k] = keyOwner + continue + } + + if hasSubjectOwner { + resolved[k] = subjectOwner + } + } + + return resolved, ambiguous } // UpdateCustomer updates a customer diff --git a/openmeter/customer/service/customer_test.go b/openmeter/customer/service/customer_test.go new file mode 100644 index 0000000000..df3e06391d --- /dev/null +++ b/openmeter/customer/service/customer_test.go @@ -0,0 +1,116 @@ +package customerservice + +import ( + "testing" + + "github.com/samber/lo" + "github.com/stretchr/testify/assert" + + "github.com/openmeterio/openmeter/openmeter/customer" + "github.com/openmeterio/openmeter/pkg/models" +) + +func Test_resolveCustomersByKey(t *testing.T) { + customerA := customer.Customer{ + ManagedResource: models.ManagedResource{ID: "customer-a"}, + Key: lo.ToPtr("key-a"), + UsageAttribution: &customer.CustomerUsageAttribution{ + SubjectKeys: []string{"subject-a"}, + }, + } + + t.Run("MatchByKey", func(t *testing.T) { + resolved, ambiguous := resolveCustomersByKey([]customer.Customer{customerA}, []string{"key-a"}) + + assert.Equal(t, map[string]customer.Customer{"key-a": customerA}, resolved) + assert.Empty(t, ambiguous) + }) + + t.Run("MatchBySubject", func(t *testing.T) { + resolved, ambiguous := resolveCustomersByKey([]customer.Customer{customerA}, []string{"subject-a"}) + + assert.Equal(t, map[string]customer.Customer{"subject-a": customerA}, resolved) + assert.Empty(t, ambiguous) + }) + + t.Run("UnmatchedKeyIsAbsent", func(t *testing.T) { + resolved, ambiguous := resolveCustomersByKey([]customer.Customer{customerA}, []string{"no-such-key"}) + + assert.Empty(t, resolved) + assert.Empty(t, ambiguous) + }) + + t.Run("KeyOwnerTakesPrecedenceOverDistinctSubjectOwner", func(t *testing.T) { + // "shared" is customerA's own key AND customerB's subject key: the key-owner (A) must win, + // mirroring the single-key GetCustomerByUsageAttribution UNION ALL lookup_priority ordering. + sharedKeyCustomer := customer.Customer{ + ManagedResource: models.ManagedResource{ID: "customer-a"}, + Key: lo.ToPtr("shared"), + } + sharedSubjectCustomer := customer.Customer{ + ManagedResource: models.ManagedResource{ID: "customer-b"}, + UsageAttribution: &customer.CustomerUsageAttribution{ + SubjectKeys: []string{"shared"}, + }, + } + + resolved, ambiguous := resolveCustomersByKey( + []customer.Customer{sharedKeyCustomer, sharedSubjectCustomer}, + []string{"shared"}, + ) + + assert.Equal(t, map[string]customer.Customer{"shared": sharedKeyCustomer}, resolved) + assert.Equal(t, []ambiguousUsageAttributionMatch{ + {Key: "shared", KeyOwnerID: "customer-a", SubjectOwnerID: "customer-b"}, + }, ambiguous) + }) + + t.Run("NoCollisionCountedWhenKeyAndSubjectOwnerAreTheSameCustomer", func(t *testing.T) { + selfMatched := customer.Customer{ + ManagedResource: models.ManagedResource{ID: "customer-a"}, + Key: lo.ToPtr("dual"), + UsageAttribution: &customer.CustomerUsageAttribution{ + SubjectKeys: []string{"dual"}, + }, + } + + resolved, ambiguous := resolveCustomersByKey([]customer.Customer{selfMatched}, []string{"dual"}) + + assert.Equal(t, map[string]customer.Customer{"dual": selfMatched}, resolved) + assert.Empty(t, ambiguous) + }) + + t.Run("CrossCollisionResolvesEachKeyToItsOwnKeyOwner", func(t *testing.T) { + // customerA's key is customerB's subject key and vice versa: no single ordering of + // [A, B] can satisfy both keys under a naive single-pass first-match-wins map, but + // resolving key-owner and subject-owner independently per key does. + crossA := customer.Customer{ + ManagedResource: models.ManagedResource{ID: "customer-a"}, + Key: lo.ToPtr("key-1"), + UsageAttribution: &customer.CustomerUsageAttribution{ + SubjectKeys: []string{"key-2"}, + }, + } + crossB := customer.Customer{ + ManagedResource: models.ManagedResource{ID: "customer-b"}, + Key: lo.ToPtr("key-2"), + UsageAttribution: &customer.CustomerUsageAttribution{ + SubjectKeys: []string{"key-1"}, + }, + } + + resolved, ambiguous := resolveCustomersByKey( + []customer.Customer{crossA, crossB}, + []string{"key-1", "key-2"}, + ) + + assert.Equal(t, map[string]customer.Customer{ + "key-1": crossA, + "key-2": crossB, + }, resolved) + assert.Equal(t, []ambiguousUsageAttributionMatch{ + {Key: "key-1", KeyOwnerID: "customer-a", SubjectOwnerID: "customer-b"}, + {Key: "key-2", KeyOwnerID: "customer-b", SubjectOwnerID: "customer-a"}, + }, ambiguous) + }) +} diff --git a/openmeter/customer/service/hooks/subjectcustomer_test.go b/openmeter/customer/service/hooks/subjectcustomer_test.go index 5e3dfd71f4..9df6c26f44 100644 --- a/openmeter/customer/service/hooks/subjectcustomer_test.go +++ b/openmeter/customer/service/hooks/subjectcustomer_test.go @@ -337,6 +337,7 @@ func NewTestEnv(t *testing.T) *TestEnv { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, + Logger: logger, }) require.NoErrorf(t, err, "initializing subject service must not fail") require.NotNilf(t, subjectAdapter, "subject service must not be nil") diff --git a/openmeter/customer/service/service.go b/openmeter/customer/service/service.go index 1ebdcc0396..b7288e0cc0 100644 --- a/openmeter/customer/service/service.go +++ b/openmeter/customer/service/service.go @@ -2,6 +2,7 @@ package customerservice import ( "errors" + "log/slog" "github.com/openmeterio/openmeter/openmeter/customer" "github.com/openmeterio/openmeter/openmeter/watermill/eventbus" @@ -14,6 +15,7 @@ type Service struct { adapter customer.Adapter requestValidatorRegistry customer.RequestValidatorRegistry publisher eventbus.Publisher + logger *slog.Logger hooks models.ServiceHookRegistry[customer.Customer] } @@ -25,6 +27,7 @@ func (s *Service) RegisterHooks(hooks ...models.ServiceHook[customer.Customer]) type Config struct { Adapter customer.Adapter Publisher eventbus.Publisher + Logger *slog.Logger } func (c Config) Validate() error { @@ -36,6 +39,10 @@ func (c Config) Validate() error { return errors.New("publisher cannot be null") } + if c.Logger == nil { + return errors.New("logger cannot be null") + } + return nil } @@ -48,6 +55,7 @@ func New(config Config) (*Service, error) { adapter: config.Adapter, requestValidatorRegistry: customer.NewRequestValidatorRegistry(), publisher: config.Publisher, + logger: config.Logger, hooks: models.ServiceHookRegistry[customer.Customer]{}, }, nil } diff --git a/openmeter/customer/testutils/env.go b/openmeter/customer/testutils/env.go index 3314df6cd5..04e25ca44e 100644 --- a/openmeter/customer/testutils/env.go +++ b/openmeter/customer/testutils/env.go @@ -105,6 +105,7 @@ func NewTestEnv(t *testing.T) *TestEnv { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, + Logger: logger, }) require.NoErrorf(t, err, "initializing subject service must not fail") require.NotNilf(t, customerService, "subject service must not be nil") diff --git a/openmeter/entitlement/metered/lateevents_test.go b/openmeter/entitlement/metered/lateevents_test.go index 0ad2d9b962..a835d1564a 100644 --- a/openmeter/entitlement/metered/lateevents_test.go +++ b/openmeter/entitlement/metered/lateevents_test.go @@ -158,6 +158,7 @@ func TestGetEntitlementBalanceConsistency(t *testing.T) { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: mockPublisher, + Logger: testLogger, }) require.NoError(t, err) diff --git a/openmeter/entitlement/metered/utils_test.go b/openmeter/entitlement/metered/utils_test.go index 1d31c80770..04a4dad2d1 100644 --- a/openmeter/entitlement/metered/utils_test.go +++ b/openmeter/entitlement/metered/utils_test.go @@ -139,6 +139,7 @@ func setupConnector(t *testing.T) (meteredentitlement.Connector, *dependencies) customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: mockPublisher, + Logger: testLogger, }) require.NoError(t, err) diff --git a/openmeter/entitlement/service/utils_test.go b/openmeter/entitlement/service/utils_test.go index 26fe81dc81..cea0df2b52 100644 --- a/openmeter/entitlement/service/utils_test.go +++ b/openmeter/entitlement/service/utils_test.go @@ -134,6 +134,7 @@ func setupDependecies(t *testing.T) (entitlement.Service, *dependencies) { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: eventbus.NewMock(t), + Logger: testLogger, }) if err != nil { t.Fatalf("failed to create customer service: %v", err) diff --git a/openmeter/governance/service/service.go b/openmeter/governance/service/service.go index 209e3868c8..74fe9ebb34 100644 --- a/openmeter/governance/service/service.go +++ b/openmeter/governance/service/service.go @@ -287,7 +287,10 @@ func (s *service) resolveCustomers(ctx context.Context, input governance.QueryAc attribute.Int("requested", len(input.CustomerKeys)), ) - customers, err := s.customerService.GetCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ + // GetCustomersByUsageAttribution already resolves each key to its customer with + // key-over-subject precedence applied, so there is no local mapping/collision handling + // left to do here. + keyToCustomer, err := s.customerService.GetCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ Namespace: input.Namespace, Keys: input.CustomerKeys, }) @@ -295,29 +298,6 @@ func (s *service) resolveCustomers(ctx context.Context, input governance.QueryAc return resolveCustomersResult{}, fmt.Errorf("failed to resolve customer keys: %w", err) } - // Map each lookup key to the customer it resolved to. A key matches a customer either by - // the customer's own key or by one of its subject keys. First match wins on the rare - // collision, mirroring the single-key First() lookup. - keyToCustomer := make(map[string]*customer.Customer, len(customers)) - - for i := range customers { - cus := &customers[i] - - if cus.Key != nil { - if _, ok := keyToCustomer[*cus.Key]; !ok { - keyToCustomer[*cus.Key] = cus - } - } - - if cus.UsageAttribution != nil { - for _, sk := range cus.UsageAttribution.SubjectKeys { - if _, ok := keyToCustomer[sk]; !ok { - keyToCustomer[sk] = cus - } - } - } - } - customerMap := make(map[string]*resolvedCustomer) var queryErrors []governance.QueryError @@ -338,7 +318,7 @@ func (s *service) resolveCustomers(ctx context.Context, input governance.QueryAc rc.matched = append(rc.matched, key) } else { customerMap[cus.ID] = &resolvedCustomer{ - customer: *cus, + customer: cus, matched: []string{key}, } } diff --git a/openmeter/governance/service/service_test.go b/openmeter/governance/service/service_test.go index 81352f6f3d..2c46ae5c19 100644 --- a/openmeter/governance/service/service_test.go +++ b/openmeter/governance/service/service_test.go @@ -91,6 +91,7 @@ func setupTestDeps(t *testing.T) *testDeps { customerSvc, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: eventbus.NewMock(t), + Logger: logger, }) require.NoError(t, err) diff --git a/openmeter/server/server_test.go b/openmeter/server/server_test.go index 2438a81961..67a9112b76 100644 --- a/openmeter/server/server_test.go +++ b/openmeter/server/server_test.go @@ -1458,7 +1458,7 @@ func (n NoopCustomerService) GetCustomerByUsageAttribution(ctx context.Context, return &customer.Customer{}, nil } -func (n NoopCustomerService) GetCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) ([]customer.Customer, error) { +func (n NoopCustomerService) GetCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) (map[string]customer.Customer, error) { return nil, nil } diff --git a/openmeter/subject/service/hooks/customersubject_test.go b/openmeter/subject/service/hooks/customersubject_test.go index c1051a08bb..0e910ee037 100644 --- a/openmeter/subject/service/hooks/customersubject_test.go +++ b/openmeter/subject/service/hooks/customersubject_test.go @@ -289,6 +289,7 @@ func NewTestEnv(t *testing.T) *TestEnv { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, + Logger: logger, }) require.NoErrorf(t, err, "initializing subject service must not fail") require.NotNilf(t, subjectAdapter, "subject service must not be nil") diff --git a/openmeter/subject/testutils/env.go b/openmeter/subject/testutils/env.go index 0c78dd3892..9d95fbfcc0 100644 --- a/openmeter/subject/testutils/env.go +++ b/openmeter/subject/testutils/env.go @@ -119,6 +119,7 @@ func NewTestEnv(t *testing.T) *TestEnv { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, + Logger: logger, }) require.NoErrorf(t, err, "initializing subject service must not fail") require.NotNilf(t, customerService, "subject service must not be nil") diff --git a/openmeter/subscription/testutils/customer.go b/openmeter/subscription/testutils/customer.go index c2c1c244f7..a491ddb798 100644 --- a/openmeter/subscription/testutils/customer.go +++ b/openmeter/subscription/testutils/customer.go @@ -63,6 +63,7 @@ func NewCustomerService(t *testing.T, dbDeps *DBDeps) customer.Service { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: eventbus.NewMock(t), + Logger: testutils.NewLogger(t), }) if err != nil { t.Fatalf("failed to create customer service: %v", err) diff --git a/test/app/stripe/testenv.go b/test/app/stripe/testenv.go index a46bfa4293..426e8d8af7 100644 --- a/test/app/stripe/testenv.go +++ b/test/app/stripe/testenv.go @@ -114,6 +114,7 @@ func NewTestEnv(t *testing.T, ctx context.Context) (TestEnv, error) { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, + Logger: logger, }) if err != nil { return nil, fmt.Errorf("failed to create customer service: %w", err) diff --git a/test/app/testenv.go b/test/app/testenv.go index 8f50fb17a1..9266b1f329 100644 --- a/test/app/testenv.go +++ b/test/app/testenv.go @@ -96,6 +96,7 @@ func NewTestEnv(t *testing.T, ctx context.Context) (TestEnv, error) { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, + Logger: logger, }) if err != nil { return nil, fmt.Errorf("failed to create customer service: %w", err) diff --git a/test/billing/suite.go b/test/billing/suite.go index 60cdf2aff1..cc2548d28c 100644 --- a/test/billing/suite.go +++ b/test/billing/suite.go @@ -164,6 +164,7 @@ func (s *BaseSuite) setupSuite() { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, + Logger: slog.Default(), }) require.NoError(t, err) s.CustomerService = customerService diff --git a/test/customer/testenv.go b/test/customer/testenv.go index c422fd4fe3..0df41e8847 100644 --- a/test/customer/testenv.go +++ b/test/customer/testenv.go @@ -187,6 +187,7 @@ func NewTestEnv(t *testing.T, ctx context.Context) (TestEnv, error) { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, + Logger: logger, }) if err != nil { return nil, err diff --git a/test/entitlement/regression/framework_test.go b/test/entitlement/regression/framework_test.go index 3641fe2304..e07c857d6a 100644 --- a/test/entitlement/regression/framework_test.go +++ b/test/entitlement/regression/framework_test.go @@ -145,6 +145,7 @@ func setupDependencies(t *testing.T) Dependencies { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: mockPublisher, + Logger: log, }) require.NoError(t, err) From 309ceee1bb1a5ffacea8520ecf6acf05c081fa3d Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Fri, 10 Jul 2026 19:13:59 +0200 Subject: [PATCH 03/13] test(e2e): add governance namespace-scale benchmark Isolates namespace size from the customers/features axes BenchmarkGovernanceQuery already covers: fixes query load at 100 customers x 1 feature, varies decoy customers/subjects seeded directly via SQL (HTTP creation would dominate setup time at 100k+). Confirmed against the pre-fix bulk lookup: latency scales with namespace size (38ms -> 160ms at 0 -> 100k decoys); post-fix it stays flat (48ms -> 54ms), matching the adapter-level benchmark's win. Moves initE2EPostgresPool into helpers.go so both this and the existing ledger e2e check can share it. --- e2e/Makefile | 6 + e2e/governance_bench_namespace_test.go | 166 +++++++++++++++++++++++++ e2e/helpers.go | 30 +++++ e2e/ledger_accounts_test.go | 24 ---- 4 files changed, 202 insertions(+), 24 deletions(-) create mode 100644 e2e/governance_bench_namespace_test.go diff --git a/e2e/Makefile b/e2e/Makefile index 09b94b15ac..26de6256b8 100644 --- a/e2e/Makefile +++ b/e2e/Makefile @@ -22,6 +22,12 @@ bench-governance-matrix: ## Benchmark governance query latency (full 3x3 custome TZ=UTC OPENMETER_ADDRESS=$(OPENMETER_ADDRESS) GOV_BENCH_FULL_MATRIX=1 \ go test -run='^$$' -bench='^BenchmarkGovernanceQuery$$' -benchmem -benchtime=$(BENCHTIME) -count=$(COUNT) ./... +.PHONY: bench-governance-namespace +bench-governance-namespace: ## Benchmark governance query latency vs. namespace size (fixed 100 customers x 1 feature, 0/10k/100k decoys); requires direct Postgres access + $(call print-target) + TZ=UTC OPENMETER_ADDRESS=$(OPENMETER_ADDRESS) \ + go test -run='^$$' -bench='^BenchmarkGovernanceQueryNamespaceScale$$' -benchmem -benchtime=$(BENCHTIME) -count=$(COUNT) ./... + .PHONY: test-local test-local: ## Run tests against local openmeter $(call print-target) diff --git a/e2e/governance_bench_namespace_test.go b/e2e/governance_bench_namespace_test.go new file mode 100644 index 0000000000..b5605bf33d --- /dev/null +++ b/e2e/governance_bench_namespace_test.go @@ -0,0 +1,166 @@ +package e2e + +import ( + "fmt" + "net/http" + "os" + "strconv" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/oklog/ulid/v2" + "github.com/stretchr/testify/require" + + v3sdk "github.com/openmeterio/openmeter/api/v3/client" +) + +const ( + defaultGovernanceBenchNamespaceDecoys = 100_000 + governanceBenchNamespaceDecoysEnv = "GOV_BENCH_NAMESPACE_DECOYS" +) + +// BenchmarkGovernanceQueryNamespaceScale isolates the effect of total namespace +// size on governance query latency, independent of the customers/features axes +// BenchmarkGovernanceQuery already covers. It fixes a small, constant query +// load (100 customers, matching the OAS customer-key cap, x 1 feature) so the +// entitlement GetAccess fan-out cost — the dominant cost per +// BenchmarkGovernanceQuery's doc comment — stays flat across sub-benchmarks, +// and varies only how many OTHER customers/subjects exist in the namespace +// (decoys, never queried, seeded directly via SQL for speed, not HTTP). +// +// This targets the customer usage-attribution resolution path specifically: +// pre-UNION-ALL, a large decoy count made GetCustomersByUsageAttribution +// seq-scan the customers table (see #4684 and the follow-up bulk fix); this +// benchmark shows whether decoy count still moves total request latency +// post-fix. Decoy counts accumulate across sub-benchmarks (0 -> 10k -> N), so +// each step only seeds the incremental delta. +// +// Requires direct Postgres access alongside OPENMETER_ADDRESS (see +// initE2EPostgresPool) — creating decoys over HTTP would dominate the +// benchmark's own setup time long before showing anything about the query +// path. Set GOV_BENCH_NAMESPACE_DECOYS to change the top decoy count from the +// default 100,000. +func BenchmarkGovernanceQueryNamespaceScale(b *testing.B) { + client := initClient(b) + v3 := newV3Client(b) + pool := initE2EPostgresPool(b) + + const ( + queryCustomers = 100 + queryFeatures = 1 + ) + + custKeys, featKeys := seedGovernanceFixture(b, client, queryCustomers, queryFeatures) + namespace := getCustomerNamespaceByKey(b, pool, custKeys[0]) + decoyRun := uniqueKey("gov_bench_ns_decoy") + + decoyCounts := []int{0, 10_000, governanceBenchNamespaceDecoyCount(b)} + + reqBody := v3sdk.GovernanceQueryRequest{ + Customer: v3sdk.GovernanceQueryRequestCustomers{Keys: custKeys}, + Feature: &v3sdk.GovernanceQueryRequestFeatures{Keys: featKeys}, + } + + seeded := 0 + for _, target := range decoyCounts { + b.Run(fmt.Sprintf("decoys=%d", target), func(b *testing.B) { + if target > seeded { + seedNamespaceDecoys(b, pool, namespace, decoyRun, seeded, target) + seeded = target + } + + // Warm-up + correctness gate: a wrong result (e.g. missing customers) + // would make the latency number meaningless. + resp, err := v3.Governance.QueryAccess(b.Context(), reqBody, v3sdk.GovernanceQueryResultListParams{}) + v3.requireStatus(http.StatusOK, err) + require.Lenf(b, resp.Data, queryCustomers, "expected %d resolved customers", queryCustomers) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := v3.Governance.QueryAccess(b.Context(), reqBody, v3sdk.GovernanceQueryResultListParams{}); err != nil { + b.Fatalf("governance query failed: %v", err) + } + if s := v3.statuses.last(); s != http.StatusOK { + b.Fatalf("governance query returned %d", s) + } + } + b.StopTimer() + b.ReportMetric(float64(target), "decoys") + }) + } +} + +func governanceBenchNamespaceDecoyCount(b *testing.B) int { + b.Helper() + + value := os.Getenv(governanceBenchNamespaceDecoysEnv) + if value == "" { + return defaultGovernanceBenchNamespaceDecoys + } + + count, err := strconv.Atoi(value) + if err != nil || count < 1 { + b.Fatalf("%s must be a positive integer, got %q", governanceBenchNamespaceDecoysEnv, value) + } + + return count +} + +func getCustomerNamespaceByKey(tb testing.TB, pool *pgxpool.Pool, key string) string { + tb.Helper() + + var namespace string + err := pool.QueryRow( + tb.Context(), + `SELECT namespace FROM customers WHERE key = $1`, + key, + ).Scan(&namespace) + require.NoError(tb, err) + + return namespace +} + +// seedNamespaceDecoys bulk-inserts decoy customers (and one subject key each) +// directly via SQL, numbered (from, to] under the given per-benchmark-run +// prefix, so repeated calls with a growing `to` only seed the incremental +// delta, and separate benchmark runs never collide on the same key even if +// the database was not reset in between. Decoys are never included in a +// governance query; they exist purely to grow the namespace the resolution +// query has to search. ANALYZE keeps planner stats current, matching the +// customer adapter's own usage-attribution benchmark. +func seedNamespaceDecoys(tb testing.TB, pool *pgxpool.Pool, namespace, runPrefix string, from, to int) { + tb.Helper() + + n := to - from + ids := make([]string, n) + keys := make([]string, n) + subjectKeys := make([]string, n) + + for i := range n { + ids[i] = ulid.Make().String() + keys[i] = fmt.Sprintf("%s_%d", runPrefix, from+i) + subjectKeys[i] = keys[i] + "_subj" + } + + rctx := tb.Context() + + _, err := pool.Exec(rctx, ` + INSERT INTO customers (id, namespace, created_at, updated_at, name, key) + SELECT id, $1, now(), now(), key, key + FROM unnest($2::text[], $3::text[]) AS t(id, key) + `, namespace, ids, keys) + require.NoError(tb, err) + + _, err = pool.Exec(rctx, ` + INSERT INTO customer_subjects (namespace, subject_key, created_at, customer_id) + SELECT $1, subject_key, now(), customer_id + FROM unnest($2::text[], $3::text[]) AS t(subject_key, customer_id) + `, namespace, subjectKeys, ids) + require.NoError(tb, err) + + _, err = pool.Exec(rctx, "ANALYZE customers") + require.NoError(tb, err) + _, err = pool.Exec(rctx, "ANALYZE customer_subjects") + require.NoError(tb, err) +} diff --git a/e2e/helpers.go b/e2e/helpers.go index 7940807114..d1dcf6e048 100644 --- a/e2e/helpers.go +++ b/e2e/helpers.go @@ -4,14 +4,44 @@ import ( "context" "fmt" "net/http" + "os" + "strings" "testing" + "github.com/jackc/pgx/v5/pgxpool" "github.com/samber/lo" "github.com/stretchr/testify/require" api "github.com/openmeterio/openmeter/api/client/go" ) +// initE2EPostgresPool opens a direct Postgres connection alongside OPENMETER_ADDRESS, for +// e2e checks and fixtures that need to read or seed data the public API doesn't expose +// (ledger account mappings, bulk namespace seeding). Requires OPENMETER_E2E_POSTGRES_URL, +// or the local compose stack's default exposed port. +func initE2EPostgresPool(t testing.TB) *pgxpool.Pool { + t.Helper() + + dsn := os.Getenv("OPENMETER_E2E_POSTGRES_URL") + if dsn == "" { + address := os.Getenv("OPENMETER_ADDRESS") + if !strings.Contains(address, "localhost:38888") && !strings.Contains(address, "127.0.0.1:38888") { + t.Skipf("this e2e check requires OPENMETER_E2E_POSTGRES_URL or local compose stack at localhost:38888, got %q", address) + } + + dsn = "postgres://postgres:postgres@127.0.0.1:35432/postgres?sslmode=disable" + } + + pool, err := pgxpool.New(t.Context(), dsn) + require.NoError(t, err) + + t.Cleanup(pool.Close) + + require.NoError(t, pool.Ping(t.Context())) + + return pool +} + // This will not be needed once we get rid of subjects. Cloud middleware already handles subject / customer creation. func CreateCustomerWithSubject(t *testing.T, client *api.ClientWithResponses, customerKey string, subjectKey string) *api.Customer { t.Helper() diff --git a/e2e/ledger_accounts_test.go b/e2e/ledger_accounts_test.go index 24ce7267fc..da5e370391 100644 --- a/e2e/ledger_accounts_test.go +++ b/e2e/ledger_accounts_test.go @@ -1,7 +1,6 @@ package e2e import ( - "os" "strings" "testing" @@ -38,29 +37,6 @@ func TestNewCustomerHasLedgerAccounts(t *testing.T) { } } -func initE2EPostgresPool(t *testing.T) *pgxpool.Pool { - t.Helper() - - dsn := os.Getenv("OPENMETER_E2E_POSTGRES_URL") - if dsn == "" { - address := os.Getenv("OPENMETER_ADDRESS") - if !strings.Contains(address, "localhost:38888") && !strings.Contains(address, "127.0.0.1:38888") { - t.Skipf("ledger account e2e requires OPENMETER_E2E_POSTGRES_URL or local compose stack at localhost:38888, got %q", address) - } - - dsn = "postgres://postgres:postgres@127.0.0.1:35432/postgres?sslmode=disable" - } - - pool, err := pgxpool.New(t.Context(), dsn) - require.NoError(t, err) - - t.Cleanup(pool.Close) - - require.NoError(t, pool.Ping(t.Context())) - - return pool -} - func getCustomerNamespace(t *testing.T, pool *pgxpool.Pool, customerID string) string { t.Helper() From a80164a5be9b4e14d460166b3ddffa5148dc788a Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Mon, 13 Jul 2026 13:24:27 +0200 Subject: [PATCH 04/13] fix(e2e): sort decoy tiers so benchmark labels match seeded namespace size decoyCounts hardcoded 10_000 as the middle tier; GOV_BENCH_NAMESPACE_DECOYS below that seeded up to 10_000 first, then skipped seeding for the smaller tier (seeded > target) while still reporting the smaller, now-incorrect count as the benchmark label. Sort + dedup keeps the sequence monotonic regardless of the env override. --- e2e/governance_bench_namespace_test.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/e2e/governance_bench_namespace_test.go b/e2e/governance_bench_namespace_test.go index b5605bf33d..f70cb0d09b 100644 --- a/e2e/governance_bench_namespace_test.go +++ b/e2e/governance_bench_namespace_test.go @@ -4,6 +4,8 @@ import ( "fmt" "net/http" "os" + "slices" + "sort" "strconv" "testing" @@ -54,7 +56,14 @@ func BenchmarkGovernanceQueryNamespaceScale(b *testing.B) { namespace := getCustomerNamespaceByKey(b, pool, custKeys[0]) decoyRun := uniqueKey("gov_bench_ns_decoy") + // Sorted and deduplicated so the cumulative seeding below stays monotonically + // increasing regardless of GOV_BENCH_NAMESPACE_DECOYS: an override below the + // hardcoded 10_000 tier would otherwise seed 10_000 first, then skip seeding + // for the smaller tier (seeded > target) while still reporting that smaller, + // now-incorrect decoy count as the benchmark's label. decoyCounts := []int{0, 10_000, governanceBenchNamespaceDecoyCount(b)} + sort.Ints(decoyCounts) + decoyCounts = slices.Compact(decoyCounts) reqBody := v3sdk.GovernanceQueryRequest{ Customer: v3sdk.GovernanceQueryRequestCustomers{Keys: custKeys}, From 5c74c6a7bb66c464aac890e29bf12f99bf39c80f Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Mon, 13 Jul 2026 20:47:40 +0200 Subject: [PATCH 05/13] fix(customer): drop unreachable created_at guard from bulk usage attribution lookup Bulk subject-key matching added a CreatedAtLTE guard the single-key path never had, diverging from its own eager-loaded subject list and forcing a second query to reconcile them. created_at is server-assigned and immutable, so a future-created subject can't occur in production. Remove the guard to restore parity with the single-key predicate. --- openmeter/customer/adapter/customer.go | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/openmeter/customer/adapter/customer.go b/openmeter/customer/adapter/customer.go index 7334a01752..1a93be51fb 100644 --- a/openmeter/customer/adapter/customer.go +++ b/openmeter/customer/adapter/customer.go @@ -631,7 +631,6 @@ func customerMatchesUsageAttributionKey(namespace, key string, at time.Time) pre // FROM customer_subjects AS cs // JOIN customers AS c ON c.id = cs.customer_id // WHERE cs.namespace = $1 AND cs.subject_key IN (...) -// AND cs.created_at <= $2 // AND (cs.deleted_at IS NULL OR cs.deleted_at > $2) // AND c.namespace = $1 AND c.deleted_at IS NULL // ) @@ -660,7 +659,6 @@ func customersMatchUsageAttributionKeys(namespace string, keys []string, at time Where(sql.And( sql.EQ(customerSubjectsTable.C(customersubjectsdb.FieldNamespace), namespace), sql.In(customerSubjectsTable.C(customersubjectsdb.FieldSubjectKey), keyValues...), - sql.LTE(customerSubjectsTable.C(customersubjectsdb.FieldCreatedAt), at), sql.Or( sql.IsNull(customerSubjectsTable.C(customersubjectsdb.FieldDeletedAt)), sql.GT(customerSubjectsTable.C(customersubjectsdb.FieldDeletedAt), at), @@ -683,8 +681,15 @@ func (a *adapter) GetCustomersByUsageAttribution(ctx context.Context, input cust return entutils.TransactingRepo(ctx, a, func(ctx context.Context, repo *adapter) ([]customer.Customer, error) { now := clock.Now().UTC() - // TODO: consider adding a CreatedAtLTE(now) guard to the customer/subject query, - // to defend against the edge case of having customers/subjects becoming active in the future + // No CreatedAtLTE(now) guard here or in WithSubjects below, deliberately: this keeps the + // candidate predicate and the eager-loaded subject list in agreement (both consider every + // subject regardless of created_at), matching the single-key + // customerMatchesUsageAttributionKey, which has never had this guard. created_at is + // server-assigned, immutable, and never caller-supplied (see the ResourceMixin/ + // CustomerSubjects schema), so a future-created subject cannot occur in production; adding + // the guard to only one of the two paths (candidate query vs. eager load) would instead + // introduce a real, batch-composition-dependent resolution bug in the caller's per-key + // precedence logic for a case that can't happen. query := repo.db.Customer.Query(). Where( customerdb.Namespace(input.Namespace), @@ -692,8 +697,6 @@ func (a *adapter) GetCustomersByUsageAttribution(ctx context.Context, input cust customersMatchUsageAttributionKeys(input.Namespace, input.Keys, now), ) - // TODO: consider adding a CreatedAtLTE(now) guard to the WithSubjects query, - // to defend against the edge case of having customers/subjects becoming active in the future query = WithSubjects(query, now) if slices.Contains(input.Expands, customer.ExpandSubscriptions) { From 9ed2ef6da65c613614831232a893d55f132369e0 Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Mon, 13 Jul 2026 21:11:56 +0200 Subject: [PATCH 06/13] docs(customer): simplify comments added on this branch Trim redundant clauses in two comments (governance namespace-scale benchmark doc, bulk usage-attribution created_at-guard note) that restated the same point twice or over-qualified it. No behavior change. --- e2e/governance_bench_namespace_test.go | 28 ++++++++++++-------------- openmeter/customer/adapter/customer.go | 16 +++++++-------- 2 files changed, 20 insertions(+), 24 deletions(-) diff --git a/e2e/governance_bench_namespace_test.go b/e2e/governance_bench_namespace_test.go index f70cb0d09b..a743a81db2 100644 --- a/e2e/governance_bench_namespace_test.go +++ b/e2e/governance_bench_namespace_test.go @@ -25,23 +25,21 @@ const ( // size on governance query latency, independent of the customers/features axes // BenchmarkGovernanceQuery already covers. It fixes a small, constant query // load (100 customers, matching the OAS customer-key cap, x 1 feature) so the -// entitlement GetAccess fan-out cost — the dominant cost per -// BenchmarkGovernanceQuery's doc comment — stays flat across sub-benchmarks, -// and varies only how many OTHER customers/subjects exist in the namespace -// (decoys, never queried, seeded directly via SQL for speed, not HTTP). +// entitlement GetAccess fan-out cost stays flat across sub-benchmarks, and +// varies only how many other (never-queried) customers/subjects exist in the +// namespace. // -// This targets the customer usage-attribution resolution path specifically: -// pre-UNION-ALL, a large decoy count made GetCustomersByUsageAttribution -// seq-scan the customers table (see #4684 and the follow-up bulk fix); this -// benchmark shows whether decoy count still moves total request latency -// post-fix. Decoy counts accumulate across sub-benchmarks (0 -> 10k -> N), so -// each step only seeds the incremental delta. +// This targets the customer usage-attribution resolution path: pre-UNION-ALL, +// a large decoy count made GetCustomersByUsageAttribution seq-scan the +// customers table (see #4684 and the follow-up bulk fix); this benchmark shows +// whether decoy count still moves total request latency post-fix. Decoy +// counts accumulate across sub-benchmarks (0 -> 10k -> N), so each step only +// seeds the incremental delta. // -// Requires direct Postgres access alongside OPENMETER_ADDRESS (see -// initE2EPostgresPool) — creating decoys over HTTP would dominate the -// benchmark's own setup time long before showing anything about the query -// path. Set GOV_BENCH_NAMESPACE_DECOYS to change the top decoy count from the -// default 100,000. +// Decoys are seeded directly via SQL (not HTTP, which would dominate setup +// time) and require direct Postgres access alongside OPENMETER_ADDRESS (see +// initE2EPostgresPool). Set GOV_BENCH_NAMESPACE_DECOYS to change the top +// decoy count from the default 100,000. func BenchmarkGovernanceQueryNamespaceScale(b *testing.B) { client := initClient(b) v3 := newV3Client(b) diff --git a/openmeter/customer/adapter/customer.go b/openmeter/customer/adapter/customer.go index 1a93be51fb..a83a39a955 100644 --- a/openmeter/customer/adapter/customer.go +++ b/openmeter/customer/adapter/customer.go @@ -681,15 +681,13 @@ func (a *adapter) GetCustomersByUsageAttribution(ctx context.Context, input cust return entutils.TransactingRepo(ctx, a, func(ctx context.Context, repo *adapter) ([]customer.Customer, error) { now := clock.Now().UTC() - // No CreatedAtLTE(now) guard here or in WithSubjects below, deliberately: this keeps the - // candidate predicate and the eager-loaded subject list in agreement (both consider every - // subject regardless of created_at), matching the single-key - // customerMatchesUsageAttributionKey, which has never had this guard. created_at is - // server-assigned, immutable, and never caller-supplied (see the ResourceMixin/ - // CustomerSubjects schema), so a future-created subject cannot occur in production; adding - // the guard to only one of the two paths (candidate query vs. eager load) would instead - // introduce a real, batch-composition-dependent resolution bug in the caller's per-key - // precedence logic for a case that can't happen. + // Deliberately no CreatedAtLTE(now) guard here or in WithSubjects below: this keeps the + // candidate predicate and the eager-loaded subject list in agreement, matching the single-key + // customerMatchesUsageAttributionKey, which never had this guard either. created_at is + // server-assigned and immutable (see ResourceMixin/CustomerSubjects schema), so a + // future-created subject can't occur in production — guarding only one of the two paths + // would instead cause a real, batch-composition-dependent resolution bug for a case that + // can't happen. query := repo.db.Customer.Query(). Where( customerdb.Namespace(input.Namespace), From f853f27ba68c437049bdf5494fcdba688e860307 Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Thu, 16 Jul 2026 17:16:10 +0200 Subject: [PATCH 07/13] fix(customer): apply deleted_at grace window uniformly in usage-attribution lookups Give the subject-branch owning-customer check the same Or(IsNull, GT(at)) grace window as the key branch in both GetCustomerByUsageAttribution and GetCustomersByUsageAttribution, so a customer resolves identically via its own key or a subject key. Behavior-equivalent in production (deletion always stamps clock.Now().UTC()); adds CustomerDeletedInFutureIncluded tests. --- openmeter/customer/adapter/customer.go | 36 ++++++++---- openmeter/customer/adapter/customer_test.go | 62 +++++++++++++++++++++ 2 files changed, 87 insertions(+), 11 deletions(-) diff --git a/openmeter/customer/adapter/customer.go b/openmeter/customer/adapter/customer.go index a83a39a955..c890ef13b6 100644 --- a/openmeter/customer/adapter/customer.go +++ b/openmeter/customer/adapter/customer.go @@ -514,7 +514,6 @@ func (a *adapter) GetCustomerByUsageAttribution(ctx context.Context, input custo query := repo.db.Customer.Query(). Where( customerdb.Namespace(input.Namespace), - customerdb.DeletedAtIsNil(), customerMatchesUsageAttributionKey(input.Namespace, input.Key, now), ) query = WithSubjects(query, now) @@ -552,7 +551,7 @@ func (a *adapter) GetCustomerByUsageAttribution(ctx context.Context, input custo // FROM customers AS c // WHERE c.namespace = $1 // AND c.key = $2 -// AND c.deleted_at IS NULL +// AND (c.deleted_at IS NULL OR c.deleted_at > $3) // // UNION ALL // @@ -563,7 +562,7 @@ func (a *adapter) GetCustomerByUsageAttribution(ctx context.Context, input custo // AND cs.subject_key = $2 // AND (cs.deleted_at IS NULL OR cs.deleted_at > $3) // AND c.namespace = $1 -// AND c.deleted_at IS NULL +// AND (c.deleted_at IS NULL OR c.deleted_at > $3) // ) AS matches // ORDER BY matches.lookup_priority // LIMIT 1 @@ -577,7 +576,10 @@ func customerMatchesUsageAttributionKey(namespace, key string, at time.Time) pre Where(sql.And( sql.EQ(keyCustomerTable.C(customerdb.FieldNamespace), namespace), sql.EQ(keyCustomerTable.C(customerdb.FieldKey), key), - sql.IsNull(keyCustomerTable.C(customerdb.FieldDeletedAt)), + sql.Or( + sql.IsNull(keyCustomerTable.C(customerdb.FieldDeletedAt)), + sql.GT(keyCustomerTable.C(customerdb.FieldDeletedAt), at), + ), )) customerSubjectsTable := sql.Table(customersubjectsdb.Table).As("customer_subjects") @@ -598,7 +600,10 @@ func customerMatchesUsageAttributionKey(namespace, key string, at time.Time) pre sql.GT(customerSubjectsTable.C(customersubjectsdb.FieldDeletedAt), at), ), sql.EQ(subjectCustomerTable.C(customerdb.FieldNamespace), namespace), - sql.IsNull(subjectCustomerTable.C(customerdb.FieldDeletedAt)), + sql.Or( + sql.IsNull(subjectCustomerTable.C(customerdb.FieldDeletedAt)), + sql.GT(subjectCustomerTable.C(customerdb.FieldDeletedAt), at), + ), )) candidates := customerKeyMatch. @@ -623,16 +628,20 @@ func customerMatchesUsageAttributionKey(namespace, key string, at time.Time) pre // // WHERE customers.id IN ( // SELECT c.id FROM customers AS c -// WHERE c.namespace = $1 AND c.key IN (...) AND c.deleted_at IS NULL +// WHERE c.namespace = $1 +// AND c.key IN (...) +// AND (c.deleted_at IS NULL OR c.deleted_at > $2) // // UNION ALL // // SELECT c.id // FROM customer_subjects AS cs // JOIN customers AS c ON c.id = cs.customer_id -// WHERE cs.namespace = $1 AND cs.subject_key IN (...) +// WHERE cs.namespace = $1 +// AND cs.subject_key IN (...) // AND (cs.deleted_at IS NULL OR cs.deleted_at > $2) -// AND c.namespace = $1 AND c.deleted_at IS NULL +// AND c.namespace = $1 +// AND (c.deleted_at IS NULL OR c.deleted_at > $2) // ) func customersMatchUsageAttributionKeys(namespace string, keys []string, at time.Time) predicate.Customer { return func(s *sql.Selector) { @@ -644,7 +653,10 @@ func customersMatchUsageAttributionKeys(namespace string, keys []string, at time Where(sql.And( sql.EQ(keyCustomerTable.C(customerdb.FieldNamespace), namespace), sql.In(keyCustomerTable.C(customerdb.FieldKey), keyValues...), - sql.IsNull(keyCustomerTable.C(customerdb.FieldDeletedAt)), + sql.Or( + sql.IsNull(keyCustomerTable.C(customerdb.FieldDeletedAt)), + sql.GT(keyCustomerTable.C(customerdb.FieldDeletedAt), at), + ), )) customerSubjectsTable := sql.Table(customersubjectsdb.Table).As("customer_subjects") @@ -664,7 +676,10 @@ func customersMatchUsageAttributionKeys(namespace string, keys []string, at time sql.GT(customerSubjectsTable.C(customersubjectsdb.FieldDeletedAt), at), ), sql.EQ(subjectCustomerTable.C(customerdb.FieldNamespace), namespace), - sql.IsNull(subjectCustomerTable.C(customerdb.FieldDeletedAt)), + sql.Or( + sql.IsNull(subjectCustomerTable.C(customerdb.FieldDeletedAt)), + sql.GT(subjectCustomerTable.C(customerdb.FieldDeletedAt), at), + ), )) candidates := customerKeyMatch.UnionAll(subjectKeyMatch) @@ -691,7 +706,6 @@ func (a *adapter) GetCustomersByUsageAttribution(ctx context.Context, input cust query := repo.db.Customer.Query(). Where( customerdb.Namespace(input.Namespace), - customerdb.DeletedAtIsNil(), customersMatchUsageAttributionKeys(input.Namespace, input.Keys, now), ) diff --git a/openmeter/customer/adapter/customer_test.go b/openmeter/customer/adapter/customer_test.go index 97fae3b099..c419305436 100644 --- a/openmeter/customer/adapter/customer_test.go +++ b/openmeter/customer/adapter/customer_test.go @@ -545,6 +545,39 @@ func TestGetCustomerByUsageAttribution(t *testing.T) { assert.Equal(t, id, got.ID) }) + t.Run("CustomerDeletedInFutureIncluded", func(t *testing.T) { + // given: + // - a customer whose own deleted_at is future-dated, set directly via Ent (the adapter can + // only ever stamp clock.Now()); the customer-level analog of SubjectDeletedInFutureIncluded. + // then: + // - the customer resolves via both its own key and its subject key, because the deleted_at + // grace window (deleted_at IS NULL OR > at) now applies to the owning customer on both the + // key branch and the subject branch identically. + env := newTestEnv(t) + ns := ulid.Make().String() + id := env.seedCustomerWithKey(ns, "cust-key", "subj-1") + now := freezeTime(t, time.Now()) + + _, err := env.db.Customer.Update(). + Where( + customerdb.Namespace(ns), + customerdb.ID(id), + ). + SetDeletedAt(now.Add(time.Minute)). + Save(t.Context()) + require.NoError(t, err) + + for _, key := range []string{"cust-key", "subj-1"} { + got, err := env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ + Namespace: ns, + Key: key, + }) + require.NoErrorf(t, err, "key %q", key) + require.NotNil(t, got) + assert.Equalf(t, id, got.ID, "key %q", key) + } + }) + t.Run("NamespaceIsolation", func(t *testing.T) { env := newTestEnv(t) nsA := ulid.Make().String() @@ -664,6 +697,35 @@ func TestGetCustomersByUsageAttribution(t *testing.T) { assert.Empty(t, customerIDs(got), "soft-deleted customer must not be returned") }) + t.Run("CustomerDeletedInFutureIncluded", func(t *testing.T) { + // given: + // - a customer whose own deleted_at is future-dated, set directly via Ent; the bulk analog + // of the single-key CustomerDeletedInFutureIncluded. + // then: + // - the customer is returned once for a key set covering both its own key and its subject key, + // because the deleted_at grace window now applies to the owning customer on both branches. + env := newTestEnv(t) + ns := ulid.Make().String() + id := env.seedCustomerWithKey(ns, "key-a", "subj-a") + now := freezeTime(t, time.Now()) + + _, err := env.db.Customer.Update(). + Where( + customerdb.Namespace(ns), + customerdb.ID(id), + ). + SetDeletedAt(now.Add(time.Minute)). + Save(t.Context()) + require.NoError(t, err) + + got, err := env.adapter.GetCustomersByUsageAttribution(t.Context(), customer.GetCustomersByUsageAttributionInput{ + Namespace: ns, + Keys: []string{"key-a", "subj-a"}, + }) + require.NoError(t, err) + assert.ElementsMatch(t, []string{id}, customerIDs(got)) + }) + t.Run("SoftDeletedSubjectExcludedButCustomerKeyStillMatches", func(t *testing.T) { env := newTestEnv(t) ns := ulid.Make().String() From d86a0e7e99260058ccea33fa6ee07ccbb7c7903a Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Fri, 17 Jul 2026 15:29:58 +0200 Subject: [PATCH 08/13] refactor(customer): unify usage-attribution lookup on one predicate, resolve precedence in service Delete the single-key adapter GetCustomerByUsageAttribution and its SQL-precedence helper (lookup_priority / ORDER BY / LIMIT). Both single-key and bulk lookups now use the one customersMatchUsageAttributionKeys predicate; the single-key service method calls the bulk adapter with a one-element key set and applies key-over-subject precedence in Go via resolveCustomersByKey. The service signature and error contract are unchanged (validation / not-found-same-message / never-conflict), so no OSS or downstream consumer changes. Drop the ambiguous-match logging from both single and bulk paths: key-over-subject collisions are structurally rare and resolved deterministically, so resolveCustomersByKey now returns only the resolved map. A benchmark of the single-key path (SQL-side vs Go-side precedence) confirms it is perf-neutral, with slightly fewer allocations. --- openmeter/customer/adapter.go | 1 - openmeter/customer/adapter/customer.go | 143 +----------- openmeter/customer/adapter/customer_test.go | 221 ------------------ ...stomer_usage_attribution_benchmark_test.go | 43 +++- openmeter/customer/service/customer.go | 83 +++---- openmeter/customer/service/customer_test.go | 28 +-- openmeter/customer/service/service_test.go | 70 ++++++ 7 files changed, 166 insertions(+), 423 deletions(-) diff --git a/openmeter/customer/adapter.go b/openmeter/customer/adapter.go index e234655801..ad55f36763 100644 --- a/openmeter/customer/adapter.go +++ b/openmeter/customer/adapter.go @@ -20,7 +20,6 @@ type CustomerAdapter interface { CreateCustomer(ctx context.Context, params CreateCustomerInput) (*Customer, error) DeleteCustomer(ctx context.Context, customer DeleteCustomerInput) error GetCustomer(ctx context.Context, customer GetCustomerInput) (*Customer, error) - GetCustomerByUsageAttribution(ctx context.Context, input GetCustomerByUsageAttributionInput) (*Customer, error) GetCustomersByUsageAttribution(ctx context.Context, input GetCustomersByUsageAttributionInput) ([]Customer, error) UpdateCustomer(ctx context.Context, params UpdateCustomerInput) (*Customer, error) } diff --git a/openmeter/customer/adapter/customer.go b/openmeter/customer/adapter/customer.go index c890ef13b6..fa5c7d5a4b 100644 --- a/openmeter/customer/adapter/customer.go +++ b/openmeter/customer/adapter/customer.go @@ -500,131 +500,13 @@ func (a *adapter) GetCustomer(ctx context.Context, input customer.GetCustomerInp }) } -// GetCustomerByUsageAttribution gets a customer by usage attribution -func (a *adapter) GetCustomerByUsageAttribution(ctx context.Context, input customer.GetCustomerByUsageAttributionInput) (*customer.Customer, error) { - if err := input.Validate(); err != nil { - return nil, models.NewGenericValidationError( - fmt.Errorf("error getting customer by usage attribution: %w", err), - ) - } - - return entutils.TransactingRepo(ctx, a, func(ctx context.Context, repo *adapter) (*customer.Customer, error) { - now := clock.Now().UTC() - - query := repo.db.Customer.Query(). - Where( - customerdb.Namespace(input.Namespace), - customerMatchesUsageAttributionKey(input.Namespace, input.Key, now), - ) - query = WithSubjects(query, now) - if slices.Contains(input.Expands, customer.ExpandSubscriptions) { - query = WithActiveSubscriptions(query, now) - } - - customerEntity, err := query.First(ctx) - if err != nil { - if entdb.IsNotFound(err) { - return nil, models.NewGenericNotFoundError( - fmt.Errorf("customer with subject key %s not found in %s namespace", input.Key, input.Namespace), - ) - } - - return nil, fmt.Errorf("failed to fetch customer: %w", err) - } - - if customerEntity == nil { - return nil, fmt.Errorf("invalid query result: nil customer received") - } - - return CustomerFromDBEntity(*customerEntity, input.Expands) - }) -} - -// customerMatchesUsageAttributionKey resolves a customer key before a subject -// key while keeping both lookup branches independently indexable. The returned -// candidate is applied to the outer customer query as: -// -// WHERE customers.id IN ( -// SELECT matches.id -// FROM ( -// SELECT c.id, 0 AS lookup_priority -// FROM customers AS c -// WHERE c.namespace = $1 -// AND c.key = $2 -// AND (c.deleted_at IS NULL OR c.deleted_at > $3) -// -// UNION ALL -// -// SELECT c.id, 1 AS lookup_priority -// FROM customer_subjects AS cs -// JOIN customers AS c ON c.id = cs.customer_id -// WHERE cs.namespace = $1 -// AND cs.subject_key = $2 -// AND (cs.deleted_at IS NULL OR cs.deleted_at > $3) -// AND c.namespace = $1 -// AND (c.deleted_at IS NULL OR c.deleted_at > $3) -// ) AS matches -// ORDER BY matches.lookup_priority -// LIMIT 1 -// ) -func customerMatchesUsageAttributionKey(namespace, key string, at time.Time) predicate.Customer { - return func(s *sql.Selector) { - keyCustomerTable := sql.Table(customerdb.Table).As("customer_by_key") - customerKeyMatch := sql.Select(keyCustomerTable.C(customerdb.FieldID)). - AppendSelectExprAs(sql.Expr("0"), "lookup_priority"). - From(keyCustomerTable). - Where(sql.And( - sql.EQ(keyCustomerTable.C(customerdb.FieldNamespace), namespace), - sql.EQ(keyCustomerTable.C(customerdb.FieldKey), key), - sql.Or( - sql.IsNull(keyCustomerTable.C(customerdb.FieldDeletedAt)), - sql.GT(keyCustomerTable.C(customerdb.FieldDeletedAt), at), - ), - )) - - customerSubjectsTable := sql.Table(customersubjectsdb.Table).As("customer_subjects") - subjectCustomerTable := sql.Table(customerdb.Table).As("customer_by_subject") - subjectKeyMatch := sql.Select(subjectCustomerTable.C(customerdb.FieldID)). - AppendSelectExprAs(sql.Expr("1"), "lookup_priority"). - From(customerSubjectsTable). - Join(subjectCustomerTable). - On( - subjectCustomerTable.C(customerdb.FieldID), - customerSubjectsTable.C(customersubjectsdb.FieldCustomerID), - ). - Where(sql.And( - sql.EQ(customerSubjectsTable.C(customersubjectsdb.FieldNamespace), namespace), - sql.EQ(customerSubjectsTable.C(customersubjectsdb.FieldSubjectKey), key), - sql.Or( - sql.IsNull(customerSubjectsTable.C(customersubjectsdb.FieldDeletedAt)), - sql.GT(customerSubjectsTable.C(customersubjectsdb.FieldDeletedAt), at), - ), - sql.EQ(subjectCustomerTable.C(customerdb.FieldNamespace), namespace), - sql.Or( - sql.IsNull(subjectCustomerTable.C(customerdb.FieldDeletedAt)), - sql.GT(subjectCustomerTable.C(customerdb.FieldDeletedAt), at), - ), - )) - - candidates := customerKeyMatch. - UnionAll(subjectKeyMatch). - As("usage_attribution_matches") - preferredCustomerID := sql.Select(candidates.C(customerdb.FieldID)). - From(candidates). - OrderBy(candidates.C("lookup_priority")). - Limit(1) - - s.Where(sql.In(s.C(customerdb.FieldID), preferredCustomerID)) - } -} - -// customersMatchUsageAttributionKeys is the bulk counterpart of customerMatchesUsageAttributionKey. -// It resolves customer IDs whose own key OR one of their subject keys is in the given key set, -// via the same two independently-indexable UNION ALL branches, but returns the full candidate set -// rather than picking one winner per key: a key that matches both a customer's own key and a -// different customer's subject key legitimately resolves to two distinct customers here, and it is -// the caller's responsibility to apply key-over-subject precedence per input key (see -// customer/service.GetCustomersByUsageAttribution). The generated SQL shape mirrors: +// customersMatchUsageAttributionKeys resolves customers whose own key OR one of their subject keys +// is in the given key set, via two independently-indexable UNION ALL branches. It returns the full +// candidate set rather than one winner per key: a key that matches both a customer's own key and a +// different customer's subject key legitimately resolves to two distinct customers here, so the +// caller applies key-over-subject precedence per input key (see +// customer/service.resolveCustomersByKey). Both the single-key and bulk usage-attribution lookups +// use this one predicate. The generated SQL shape mirrors: // // WHERE customers.id IN ( // SELECT c.id FROM customers AS c @@ -688,17 +570,16 @@ func customersMatchUsageAttributionKeys(namespace string, keys []string, at time } } -// GetCustomersByUsageAttribution resolves multiple customers by usage attribution keys in a single query. -// A key matches a customer either by the customer's own key or by one of its subject keys, mirroring -// the single-key GetCustomerByUsageAttribution. Keys that match no customer are simply absent from the -// result; the caller derives which keys were not found. +// GetCustomersByUsageAttribution resolves customers by usage attribution keys in a single query. +// A key matches a customer either by the customer's own key or by one of its subject keys. It is the +// sole usage-attribution lookup for both the bulk and single-key service paths; keys that match no +// customer are simply absent from the result, and the caller derives which keys were not found. func (a *adapter) GetCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) ([]customer.Customer, error) { return entutils.TransactingRepo(ctx, a, func(ctx context.Context, repo *adapter) ([]customer.Customer, error) { now := clock.Now().UTC() // Deliberately no CreatedAtLTE(now) guard here or in WithSubjects below: this keeps the - // candidate predicate and the eager-loaded subject list in agreement, matching the single-key - // customerMatchesUsageAttributionKey, which never had this guard either. created_at is + // candidate predicate and the eager-loaded subject list in agreement. created_at is // server-assigned and immutable (see ResourceMixin/CustomerSubjects schema), so a // future-created subject can't occur in production — guarding only one of the two paths // would instead cause a real, batch-composition-dependent resolution bug for a case that diff --git a/openmeter/customer/adapter/customer_test.go b/openmeter/customer/adapter/customer_test.go index c419305436..015839fe93 100644 --- a/openmeter/customer/adapter/customer_test.go +++ b/openmeter/customer/adapter/customer_test.go @@ -389,227 +389,6 @@ func customerIDs(customers []customer.Customer) []string { }) } -func TestGetCustomerByUsageAttribution(t *testing.T) { - t.Run("MatchByCustomerKey", func(t *testing.T) { - env := newTestEnv(t) - ns := ulid.Make().String() - id := env.seedCustomerWithKey(ns, "cust-key", "subj-1") - - got, err := env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ - Namespace: ns, - Key: "cust-key", - }) - require.NoError(t, err) - require.NotNil(t, got) - assert.Equal(t, id, got.ID) - require.NotNil(t, got.UsageAttribution) - assert.Equal(t, []string{"subj-1"}, got.UsageAttribution.SubjectKeys) - }) - - t.Run("MatchBySubjectKey", func(t *testing.T) { - env := newTestEnv(t) - ns := ulid.Make().String() - id := env.seedCustomerWithKey(ns, "cust-key", "subj-1", "subj-2") - - got, err := env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ - Namespace: ns, - Key: "subj-2", - }) - require.NoError(t, err) - require.NotNil(t, got) - assert.Equal(t, id, got.ID) - require.NotNil(t, got.UsageAttribution) - assert.Equal(t, []string{"subj-1", "subj-2"}, got.UsageAttribution.SubjectKeys) - }) - - t.Run("CustomerKeyTakesPriorityOverAnotherCustomerSubjectKey", func(t *testing.T) { - env := newTestEnv(t) - ns := ulid.Make().String() - customerKeyID := env.seedCustomerWithKey(ns, "shared-key", "customer-key-subject") - _ = env.seedCustomerWithKey(ns, "subject-owner", "shared-key") - - got, err := env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ - Namespace: ns, - Key: "shared-key", - }) - require.NoError(t, err) - require.NotNil(t, got) - assert.Equal(t, customerKeyID, got.ID, "a direct customer-key match must take priority over a subject-key match") - }) - - t.Run("CustomerMatchedByOwnKeyAndSubjectKeyReturnedOnce", func(t *testing.T) { - env := newTestEnv(t) - ns := ulid.Make().String() - id := env.seedCustomerWithKey(ns, "shared-key", "shared-key") - - got, err := env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ - Namespace: ns, - Key: "shared-key", - }) - require.NoError(t, err) - require.NotNil(t, got) - assert.Equal(t, id, got.ID) - }) - - t.Run("SoftDeletedCustomerExcluded", func(t *testing.T) { - env := newTestEnv(t) - ns := ulid.Make().String() - id := env.seedCustomerWithKey(ns, "cust-key", "subj-1") - _ = freezeTime(t, time.Now()) - - require.NoError(t, env.adapter.DeleteCustomer(t.Context(), customer.DeleteCustomerInput{ - Namespace: ns, - ID: id, - })) - - for _, key := range []string{"cust-key", "subj-1"} { - _, err := env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ - Namespace: ns, - Key: key, - }) - require.Error(t, err) - assert.True(t, models.IsGenericNotFoundError(err)) - } - }) - - t.Run("SubjectDeletedInPastExcluded", func(t *testing.T) { - env := newTestEnv(t) - ns := ulid.Make().String() - id := env.seedCustomerWithKey(ns, "cust-key", "subj-1") - now := freezeTime(t, time.Now()) - - _, err := env.db.CustomerSubjects.Update(). - Where( - customersubjectsdb.Namespace(ns), - customersubjectsdb.CustomerID(id), - customersubjectsdb.SubjectKey("subj-1"), - ). - SetDeletedAt(now.Add(-time.Minute)). - Save(t.Context()) - require.NoError(t, err) - - _, err = env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ - Namespace: ns, - Key: "subj-1", - }) - require.Error(t, err) - assert.True(t, models.IsGenericNotFoundError(err)) - }) - - t.Run("SubjectDeletedAtLookupTimeExcluded", func(t *testing.T) { - env := newTestEnv(t) - ns := ulid.Make().String() - id := env.seedCustomerWithKey(ns, "cust-key", "subj-1") - now := freezeTime(t, time.Now()) - - _, err := env.db.CustomerSubjects.Update(). - Where( - customersubjectsdb.Namespace(ns), - customersubjectsdb.CustomerID(id), - customersubjectsdb.SubjectKey("subj-1"), - ). - SetDeletedAt(now). - Save(t.Context()) - require.NoError(t, err) - - _, err = env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ - Namespace: ns, - Key: "subj-1", - }) - require.Error(t, err) - assert.True(t, models.IsGenericNotFoundError(err)) - }) - - t.Run("SubjectDeletedInFutureIncluded", func(t *testing.T) { - env := newTestEnv(t) - ns := ulid.Make().String() - id := env.seedCustomerWithKey(ns, "cust-key", "subj-1") - now := freezeTime(t, time.Now()) - - _, err := env.db.CustomerSubjects.Update(). - Where( - customersubjectsdb.Namespace(ns), - customersubjectsdb.CustomerID(id), - customersubjectsdb.SubjectKey("subj-1"), - ). - SetDeletedAt(now.Add(time.Minute)). - Save(t.Context()) - require.NoError(t, err) - - got, err := env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ - Namespace: ns, - Key: "subj-1", - }) - require.NoError(t, err) - require.NotNil(t, got) - assert.Equal(t, id, got.ID) - }) - - t.Run("CustomerDeletedInFutureIncluded", func(t *testing.T) { - // given: - // - a customer whose own deleted_at is future-dated, set directly via Ent (the adapter can - // only ever stamp clock.Now()); the customer-level analog of SubjectDeletedInFutureIncluded. - // then: - // - the customer resolves via both its own key and its subject key, because the deleted_at - // grace window (deleted_at IS NULL OR > at) now applies to the owning customer on both the - // key branch and the subject branch identically. - env := newTestEnv(t) - ns := ulid.Make().String() - id := env.seedCustomerWithKey(ns, "cust-key", "subj-1") - now := freezeTime(t, time.Now()) - - _, err := env.db.Customer.Update(). - Where( - customerdb.Namespace(ns), - customerdb.ID(id), - ). - SetDeletedAt(now.Add(time.Minute)). - Save(t.Context()) - require.NoError(t, err) - - for _, key := range []string{"cust-key", "subj-1"} { - got, err := env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ - Namespace: ns, - Key: key, - }) - require.NoErrorf(t, err, "key %q", key) - require.NotNil(t, got) - assert.Equalf(t, id, got.ID, "key %q", key) - } - }) - - t.Run("NamespaceIsolation", func(t *testing.T) { - env := newTestEnv(t) - nsA := ulid.Make().String() - nsB := ulid.Make().String() - idA := env.seedCustomerWithKey(nsA, "shared-customer-key", "shared-subject-key") - _ = env.seedCustomerWithKey(nsB, "shared-customer-key", "shared-subject-key") - - for _, key := range []string{"shared-customer-key", "shared-subject-key"} { - got, err := env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ - Namespace: nsA, - Key: key, - }) - require.NoError(t, err) - require.NotNil(t, got) - assert.Equal(t, idA, got.ID) - } - }) - - t.Run("UnknownKeyReturnsNotFound", func(t *testing.T) { - env := newTestEnv(t) - ns := ulid.Make().String() - _ = env.seedCustomerWithKey(ns, "cust-key", "subj-1") - - _, err := env.adapter.GetCustomerByUsageAttribution(t.Context(), customer.GetCustomerByUsageAttributionInput{ - Namespace: ns, - Key: "missing", - }) - require.Error(t, err) - assert.True(t, models.IsGenericNotFoundError(err)) - }) -} - func TestGetCustomersByUsageAttribution(t *testing.T) { t.Run("MatchByCustomerKey", func(t *testing.T) { env := newTestEnv(t) diff --git a/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go b/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go index 8d97c3a329..669bbcd779 100644 --- a/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go +++ b/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "os" + "slices" "strconv" "testing" "time" @@ -84,12 +85,50 @@ func BenchmarkCustomerUsageAttributionLookup(b *testing.B) { return client.Customer.Query(). Where( customerdb.Namespace(namespace), - customerdb.DeletedAtIsNil(), - customerMatchesUsageAttributionKey(namespace, lookup.key, now), + customersMatchUsageAttributionKeys(namespace, []string{lookup.key}, now), ). FirstID(ctx) }, }, + { + // method measures the FULL single-key path (merged predicate + WithSubjects + // hydration + domain mapping + Go-side precedence), not just the ID query, so it + // reflects the real customer/service.GetCustomerByUsageAttribution cost. The + // inline resolution mirrors service.resolveCustomersByKey over the raw candidate + // set (<=2 rows): a customer-key match wins over a subject-key match. + name: "method", + query: func(ctx context.Context) (string, error) { + q := client.Customer.Query().Where( + customerdb.Namespace(namespace), + customersMatchUsageAttributionKeys(namespace, []string{lookup.key}, now), + ) + q = WithSubjects(q, now) + entities, err := q.All(ctx) + if err != nil { + return "", err + } + var byKeyID, bySubjectID string + for _, entity := range entities { + c, err := CustomerFromDBEntity(*entity, nil) + if err != nil { + return "", err + } + if c.Key != nil && *c.Key == lookup.key { + byKeyID = c.ID + } + if c.UsageAttribution != nil && bySubjectID == "" && slices.Contains(c.UsageAttribution.SubjectKeys, lookup.key) { + bySubjectID = c.ID + } + } + if byKeyID != "" { + return byKeyID, nil + } + if bySubjectID != "" { + return bySubjectID, nil + } + return "", fmt.Errorf("no customer resolved for key %q", lookup.key) + }, + }, } for _, variant := range variants { diff --git a/openmeter/customer/service/customer.go b/openmeter/customer/service/customer.go index 1062ae5297..cf26a20e41 100644 --- a/openmeter/customer/service/customer.go +++ b/openmeter/customer/service/customer.go @@ -131,9 +131,33 @@ func (s *Service) GetCustomer(ctx context.Context, input customer.GetCustomerInp return s.adapter.GetCustomer(ctx, input) } -// GetCustomerByUsageAttribution gets a customer by usage attribution +// GetCustomerByUsageAttribution gets a customer by usage attribution. It resolves the single key +// through the same bulk candidate query and key-over-subject precedence as +// GetCustomersByUsageAttribution, so both paths share one predicate and one precedence mechanism. func (s *Service) GetCustomerByUsageAttribution(ctx context.Context, input customer.GetCustomerByUsageAttributionInput) (*customer.Customer, error) { - return s.adapter.GetCustomerByUsageAttribution(ctx, input) + if err := input.Validate(); err != nil { + return nil, models.NewGenericValidationError( + fmt.Errorf("error getting customer by usage attribution: %w", err), + ) + } + + customers, err := s.adapter.GetCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ + Namespace: input.Namespace, + Keys: []string{input.Key}, + Expands: input.Expands, + }) + if err != nil { + return nil, err + } + + c, ok := resolveCustomersByKey(customers, []string{input.Key})[input.Key] + if !ok { + return nil, models.NewGenericNotFoundError( + fmt.Errorf("customer with subject key %s not found in %s namespace", input.Key, input.Namespace), + ) + } + + return &c, nil } // GetCustomersByUsageAttribution resolves multiple customers by usage attribution keys in a single @@ -150,41 +174,16 @@ func (s *Service) GetCustomersByUsageAttribution(ctx context.Context, input cust return nil, err } - resolved, ambiguous := resolveCustomersByKey(customers, input.Keys) - - // A key resolving to both a distinct key-owner and a distinct subject-owner is not an error — - // the key-owner wins deterministically, mirroring GetCustomerByUsageAttribution's single-row - // precedence — but the underlying data (one key doing double duty as another customer's own - // key and a different customer's subject key) is unusual enough to warrant operator attention, - // so each occurrence is logged with the key and both customer IDs for follow-up. - for _, m := range ambiguous { - s.logger.ErrorContext(ctx, "ambiguous usage attribution key: matches both a customer key and a distinct subject key", - "namespace", input.Namespace, - "key", m.Key, - "key_owner.id", m.KeyOwnerID, - "subject_owner.id", m.SubjectOwnerID, - ) - } - - return resolved, nil -} - -// ambiguousUsageAttributionMatch records a key that matched both a distinct customer-key owner and -// a distinct subject-key owner, so the caller can log the identifiers needed to investigate and fix -// the underlying data overlap. -type ambiguousUsageAttributionMatch struct { - Key string - KeyOwnerID string - SubjectOwnerID string + return resolveCustomersByKey(customers, input.Keys), nil } // resolveCustomersByKey maps each input key to the customer it matches. A key matches a customer // either by the customer's own key or by one of its subject keys; when a key matches both a -// distinct key-owner and a distinct subject-owner, the key-owner takes precedence, mirroring the -// single-key GetCustomerByUsageAttribution's UNION ALL lookup_priority ordering. Keys with no match -// are absent from the returned map. The second return value lists the keys where precedence -// actually mattered (a distinct key-owner and a distinct subject-owner both matched). -func resolveCustomersByKey(customers []customer.Customer, keys []string) (map[string]customer.Customer, []ambiguousUsageAttributionMatch) { +// distinct key-owner and a distinct subject-owner, the key-owner takes precedence. Key-over-subject +// collisions are structurally rare and resolved deterministically here, so they are not surfaced as +// errors or logs; investigate the underlying data via the database if ever needed. Keys with no +// match are absent from the returned map. +func resolveCustomersByKey(customers []customer.Customer, keys []string) map[string]customer.Customer { byKey := make(map[string]customer.Customer, len(customers)) bySubject := make(map[string]customer.Customer, len(customers)) @@ -203,31 +202,19 @@ func resolveCustomersByKey(customers []customer.Customer, keys []string) (map[st } resolved := make(map[string]customer.Customer, len(keys)) - var ambiguous []ambiguousUsageAttributionMatch for _, k := range keys { - keyOwner, hasKeyOwner := byKey[k] - subjectOwner, hasSubjectOwner := bySubject[k] - - if hasKeyOwner && hasSubjectOwner && keyOwner.ID != subjectOwner.ID { - ambiguous = append(ambiguous, ambiguousUsageAttributionMatch{ - Key: k, - KeyOwnerID: keyOwner.ID, - SubjectOwnerID: subjectOwner.ID, - }) - } - - if hasKeyOwner { + if keyOwner, ok := byKey[k]; ok { resolved[k] = keyOwner continue } - if hasSubjectOwner { + if subjectOwner, ok := bySubject[k]; ok { resolved[k] = subjectOwner } } - return resolved, ambiguous + return resolved } // UpdateCustomer updates a customer diff --git a/openmeter/customer/service/customer_test.go b/openmeter/customer/service/customer_test.go index df3e06391d..9956170d5c 100644 --- a/openmeter/customer/service/customer_test.go +++ b/openmeter/customer/service/customer_test.go @@ -20,29 +20,25 @@ func Test_resolveCustomersByKey(t *testing.T) { } t.Run("MatchByKey", func(t *testing.T) { - resolved, ambiguous := resolveCustomersByKey([]customer.Customer{customerA}, []string{"key-a"}) + resolved := resolveCustomersByKey([]customer.Customer{customerA}, []string{"key-a"}) assert.Equal(t, map[string]customer.Customer{"key-a": customerA}, resolved) - assert.Empty(t, ambiguous) }) t.Run("MatchBySubject", func(t *testing.T) { - resolved, ambiguous := resolveCustomersByKey([]customer.Customer{customerA}, []string{"subject-a"}) + resolved := resolveCustomersByKey([]customer.Customer{customerA}, []string{"subject-a"}) assert.Equal(t, map[string]customer.Customer{"subject-a": customerA}, resolved) - assert.Empty(t, ambiguous) }) t.Run("UnmatchedKeyIsAbsent", func(t *testing.T) { - resolved, ambiguous := resolveCustomersByKey([]customer.Customer{customerA}, []string{"no-such-key"}) + resolved := resolveCustomersByKey([]customer.Customer{customerA}, []string{"no-such-key"}) assert.Empty(t, resolved) - assert.Empty(t, ambiguous) }) t.Run("KeyOwnerTakesPrecedenceOverDistinctSubjectOwner", func(t *testing.T) { - // "shared" is customerA's own key AND customerB's subject key: the key-owner (A) must win, - // mirroring the single-key GetCustomerByUsageAttribution UNION ALL lookup_priority ordering. + // "shared" is customerA's own key AND customerB's subject key: the key-owner (A) must win. sharedKeyCustomer := customer.Customer{ ManagedResource: models.ManagedResource{ID: "customer-a"}, Key: lo.ToPtr("shared"), @@ -54,18 +50,15 @@ func Test_resolveCustomersByKey(t *testing.T) { }, } - resolved, ambiguous := resolveCustomersByKey( + resolved := resolveCustomersByKey( []customer.Customer{sharedKeyCustomer, sharedSubjectCustomer}, []string{"shared"}, ) assert.Equal(t, map[string]customer.Customer{"shared": sharedKeyCustomer}, resolved) - assert.Equal(t, []ambiguousUsageAttributionMatch{ - {Key: "shared", KeyOwnerID: "customer-a", SubjectOwnerID: "customer-b"}, - }, ambiguous) }) - t.Run("NoCollisionCountedWhenKeyAndSubjectOwnerAreTheSameCustomer", func(t *testing.T) { + t.Run("SameCustomerMatchedByOwnKeyAndSubjectKey", func(t *testing.T) { selfMatched := customer.Customer{ ManagedResource: models.ManagedResource{ID: "customer-a"}, Key: lo.ToPtr("dual"), @@ -74,10 +67,9 @@ func Test_resolveCustomersByKey(t *testing.T) { }, } - resolved, ambiguous := resolveCustomersByKey([]customer.Customer{selfMatched}, []string{"dual"}) + resolved := resolveCustomersByKey([]customer.Customer{selfMatched}, []string{"dual"}) assert.Equal(t, map[string]customer.Customer{"dual": selfMatched}, resolved) - assert.Empty(t, ambiguous) }) t.Run("CrossCollisionResolvesEachKeyToItsOwnKeyOwner", func(t *testing.T) { @@ -99,7 +91,7 @@ func Test_resolveCustomersByKey(t *testing.T) { }, } - resolved, ambiguous := resolveCustomersByKey( + resolved := resolveCustomersByKey( []customer.Customer{crossA, crossB}, []string{"key-1", "key-2"}, ) @@ -108,9 +100,5 @@ func Test_resolveCustomersByKey(t *testing.T) { "key-1": crossA, "key-2": crossB, }, resolved) - assert.Equal(t, []ambiguousUsageAttributionMatch{ - {Key: "key-1", KeyOwnerID: "customer-a", SubjectOwnerID: "customer-b"}, - {Key: "key-2", KeyOwnerID: "customer-b", SubjectOwnerID: "customer-a"}, - }, ambiguous) }) } diff --git a/openmeter/customer/service/service_test.go b/openmeter/customer/service/service_test.go index 2fe58607fc..f7c853bd6b 100644 --- a/openmeter/customer/service/service_test.go +++ b/openmeter/customer/service/service_test.go @@ -241,3 +241,73 @@ func Test_GetCustomersByUsageAttribution_Validation(t *testing.T) { require.Error(t, err, "empty key set must fail validation") assert.True(t, models.IsGenericValidationError(err), "error must be a validation error") } + +// Test_GetCustomerByUsageAttribution covers the single-key wiring that Test_CustomerService's +// happy-path lookups don't reach: the key-over-subject precedence now resolved in the service (not +// in SQL), the not-found mapping, and input validation. The predicate itself is covered by the +// adapter's TestGetCustomersByUsageAttribution. +func Test_GetCustomerByUsageAttribution(t *testing.T) { + env := customertestutils.NewTestEnv(t) + t.Cleanup(func() { + env.Close(t) + }) + env.DBSchemaMigrate(t) + + namespace := customertestutils.NewTestNamespace(t) + ctx := t.Context() + + t.Run("KeyOverSubjectPrecedence", func(t *testing.T) { + // given: + // - customer A owns the key "shared-key" + // - customer B carries "shared-key" as one of its subject keys + // when: + // - resolving "shared-key" by usage attribution + // then: + // - the direct key owner (A) wins over the subject-key match (B), resolved in the service + keyOwner, err := env.CustomerService.CreateCustomer(ctx, customer.CreateCustomerInput{ + Namespace: namespace, + CustomerMutate: customer.CustomerMutate{ + Key: lo.ToPtr("shared-key"), + Name: "Key Owner", + UsageAttribution: &customer.CustomerUsageAttribution{SubjectKeys: []string{"key-owner-subject"}}, + }, + }) + require.NoError(t, err) + + _, err = env.CustomerService.CreateCustomer(ctx, customer.CreateCustomerInput{ + Namespace: namespace, + CustomerMutate: customer.CustomerMutate{ + Key: lo.ToPtr("subject-owner"), + Name: "Subject Owner", + UsageAttribution: &customer.CustomerUsageAttribution{SubjectKeys: []string{"shared-key"}}, + }, + }) + require.NoError(t, err) + + got, err := env.CustomerService.GetCustomerByUsageAttribution(ctx, customer.GetCustomerByUsageAttributionInput{ + Namespace: namespace, + Key: "shared-key", + }) + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, keyOwner.ID, got.ID, "a direct customer-key match must win over a subject-key match") + }) + + t.Run("UnknownKeyReturnsNotFound", func(t *testing.T) { + _, err := env.CustomerService.GetCustomerByUsageAttribution(ctx, customer.GetCustomerByUsageAttributionInput{ + Namespace: namespace, + Key: "no-such-key", + }) + require.Error(t, err) + assert.True(t, models.IsGenericNotFoundError(err), "unmatched key must map to a not-found error") + }) + + t.Run("EmptyKeyFailsValidation", func(t *testing.T) { + _, err := env.CustomerService.GetCustomerByUsageAttribution(ctx, customer.GetCustomerByUsageAttributionInput{ + Namespace: namespace, + Key: "", + }) + require.Error(t, err) + assert.True(t, models.IsGenericValidationError(err), "empty key must fail validation") + }) +} From 3e64d2f7597c593a0a16868f6b716144ece2874c Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Fri, 17 Jul 2026 15:42:09 +0200 Subject: [PATCH 09/13] test(customer): add bulk usage-attribution service tests; share resolution core Route the single-key and bulk GetCustomer(s)ByUsageAttribution service methods through one resolveCustomersByUsageAttribution helper (adapter fetch + precedence) so their core resolution can't drift. Rename resolveCustomersByKey to resolveCustomersByKeyWithPrecedence to name its key-over-subject behavior. Add Test_GetCustomersByUsageAttribution covering key/subject resolution, key-over-subject precedence, unmatched-key-absent (no error), and empty-key validation, mirroring Test_GetCustomerByUsageAttribution. The fully symmetric {K1,K2} cross-collision is left to the unit and adapter tests: CreateCustomer's overlap guard rejects a customer key that overlaps another customer's subject, so that state can't be constructed through the service. --- openmeter/customer/adapter/customer.go | 2 +- ...stomer_usage_attribution_benchmark_test.go | 2 +- openmeter/customer/service/customer.go | 17 ++- openmeter/customer/service/customer_test.go | 14 +-- openmeter/customer/service/service_test.go | 100 +++++++++++++++--- 5 files changed, 109 insertions(+), 26 deletions(-) diff --git a/openmeter/customer/adapter/customer.go b/openmeter/customer/adapter/customer.go index fa5c7d5a4b..a20a589e72 100644 --- a/openmeter/customer/adapter/customer.go +++ b/openmeter/customer/adapter/customer.go @@ -505,7 +505,7 @@ func (a *adapter) GetCustomer(ctx context.Context, input customer.GetCustomerInp // candidate set rather than one winner per key: a key that matches both a customer's own key and a // different customer's subject key legitimately resolves to two distinct customers here, so the // caller applies key-over-subject precedence per input key (see -// customer/service.resolveCustomersByKey). Both the single-key and bulk usage-attribution lookups +// customer/service.resolveCustomersByKeyWithPrecedence). Both the single-key and bulk usage-attribution lookups // use this one predicate. The generated SQL shape mirrors: // // WHERE customers.id IN ( diff --git a/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go b/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go index 669bbcd779..05a394d743 100644 --- a/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go +++ b/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go @@ -94,7 +94,7 @@ func BenchmarkCustomerUsageAttributionLookup(b *testing.B) { // method measures the FULL single-key path (merged predicate + WithSubjects // hydration + domain mapping + Go-side precedence), not just the ID query, so it // reflects the real customer/service.GetCustomerByUsageAttribution cost. The - // inline resolution mirrors service.resolveCustomersByKey over the raw candidate + // inline resolution mirrors service.resolveCustomersByKeyWithPrecedence over the raw candidate // set (<=2 rows): a customer-key match wins over a subject-key match. name: "method", query: func(ctx context.Context) (string, error) { diff --git a/openmeter/customer/service/customer.go b/openmeter/customer/service/customer.go index cf26a20e41..c5c30a6f3c 100644 --- a/openmeter/customer/service/customer.go +++ b/openmeter/customer/service/customer.go @@ -141,7 +141,7 @@ func (s *Service) GetCustomerByUsageAttribution(ctx context.Context, input custo ) } - customers, err := s.adapter.GetCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ + resolved, err := s.resolveCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ Namespace: input.Namespace, Keys: []string{input.Key}, Expands: input.Expands, @@ -150,7 +150,7 @@ func (s *Service) GetCustomerByUsageAttribution(ctx context.Context, input custo return nil, err } - c, ok := resolveCustomersByKey(customers, []string{input.Key})[input.Key] + c, ok := resolved[input.Key] if !ok { return nil, models.NewGenericNotFoundError( fmt.Errorf("customer with subject key %s not found in %s namespace", input.Key, input.Namespace), @@ -169,21 +169,28 @@ func (s *Service) GetCustomersByUsageAttribution(ctx context.Context, input cust ) } + return s.resolveCustomersByUsageAttribution(ctx, input) +} + +// resolveCustomersByUsageAttribution runs the shared usage-attribution lookup: fetch the raw +// candidate customers from the adapter, then apply key-over-subject precedence. Both the single-key +// and bulk service methods route through this so their core resolution stays identical. +func (s *Service) resolveCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) (map[string]customer.Customer, error) { customers, err := s.adapter.GetCustomersByUsageAttribution(ctx, input) if err != nil { return nil, err } - return resolveCustomersByKey(customers, input.Keys), nil + return resolveCustomersByKeyWithPrecedence(customers, input.Keys), nil } -// resolveCustomersByKey maps each input key to the customer it matches. A key matches a customer +// resolveCustomersByKeyWithPrecedence maps each input key to the customer it matches. A key matches a customer // either by the customer's own key or by one of its subject keys; when a key matches both a // distinct key-owner and a distinct subject-owner, the key-owner takes precedence. Key-over-subject // collisions are structurally rare and resolved deterministically here, so they are not surfaced as // errors or logs; investigate the underlying data via the database if ever needed. Keys with no // match are absent from the returned map. -func resolveCustomersByKey(customers []customer.Customer, keys []string) map[string]customer.Customer { +func resolveCustomersByKeyWithPrecedence(customers []customer.Customer, keys []string) map[string]customer.Customer { byKey := make(map[string]customer.Customer, len(customers)) bySubject := make(map[string]customer.Customer, len(customers)) diff --git a/openmeter/customer/service/customer_test.go b/openmeter/customer/service/customer_test.go index 9956170d5c..8c6a3952ba 100644 --- a/openmeter/customer/service/customer_test.go +++ b/openmeter/customer/service/customer_test.go @@ -10,7 +10,7 @@ import ( "github.com/openmeterio/openmeter/pkg/models" ) -func Test_resolveCustomersByKey(t *testing.T) { +func Test_resolveCustomersByKeyWithPrecedence(t *testing.T) { customerA := customer.Customer{ ManagedResource: models.ManagedResource{ID: "customer-a"}, Key: lo.ToPtr("key-a"), @@ -20,19 +20,19 @@ func Test_resolveCustomersByKey(t *testing.T) { } t.Run("MatchByKey", func(t *testing.T) { - resolved := resolveCustomersByKey([]customer.Customer{customerA}, []string{"key-a"}) + resolved := resolveCustomersByKeyWithPrecedence([]customer.Customer{customerA}, []string{"key-a"}) assert.Equal(t, map[string]customer.Customer{"key-a": customerA}, resolved) }) t.Run("MatchBySubject", func(t *testing.T) { - resolved := resolveCustomersByKey([]customer.Customer{customerA}, []string{"subject-a"}) + resolved := resolveCustomersByKeyWithPrecedence([]customer.Customer{customerA}, []string{"subject-a"}) assert.Equal(t, map[string]customer.Customer{"subject-a": customerA}, resolved) }) t.Run("UnmatchedKeyIsAbsent", func(t *testing.T) { - resolved := resolveCustomersByKey([]customer.Customer{customerA}, []string{"no-such-key"}) + resolved := resolveCustomersByKeyWithPrecedence([]customer.Customer{customerA}, []string{"no-such-key"}) assert.Empty(t, resolved) }) @@ -50,7 +50,7 @@ func Test_resolveCustomersByKey(t *testing.T) { }, } - resolved := resolveCustomersByKey( + resolved := resolveCustomersByKeyWithPrecedence( []customer.Customer{sharedKeyCustomer, sharedSubjectCustomer}, []string{"shared"}, ) @@ -67,7 +67,7 @@ func Test_resolveCustomersByKey(t *testing.T) { }, } - resolved := resolveCustomersByKey([]customer.Customer{selfMatched}, []string{"dual"}) + resolved := resolveCustomersByKeyWithPrecedence([]customer.Customer{selfMatched}, []string{"dual"}) assert.Equal(t, map[string]customer.Customer{"dual": selfMatched}, resolved) }) @@ -91,7 +91,7 @@ func Test_resolveCustomersByKey(t *testing.T) { }, } - resolved := resolveCustomersByKey( + resolved := resolveCustomersByKeyWithPrecedence( []customer.Customer{crossA, crossB}, []string{"key-1", "key-2"}, ) diff --git a/openmeter/customer/service/service_test.go b/openmeter/customer/service/service_test.go index f7c853bd6b..7fea241699 100644 --- a/openmeter/customer/service/service_test.go +++ b/openmeter/customer/service/service_test.go @@ -220,26 +220,102 @@ func Test_CustomerService(t *testing.T) { }) } -func Test_GetCustomersByUsageAttribution_Validation(t *testing.T) { +// Test_GetCustomersByUsageAttribution protects the bulk method's contract with the same rigor as +// Test_GetCustomerByUsageAttribution does the single-key one: each input key maps to its customer +// with key-over-subject precedence, and unmatched keys are simply absent (no error). The predicate +// itself is covered by the adapter's TestGetCustomersByUsageAttribution, and the fully symmetric +// {K1,K2} cross-collision by Test_resolveCustomersByKeyWithPrecedence — that state cannot be built +// through the service here, because CreateCustomer rejects a customer whose key overlaps another +// customer's subject key (see the overlap guard in the customer adapter). +func Test_GetCustomersByUsageAttribution(t *testing.T) { env := customertestutils.NewTestEnv(t) t.Cleanup(func() { env.Close(t) }) namespace := customertestutils.NewTestNamespace(t) + ctx := t.Context() + + create := func(t *testing.T, key string, subjectKeys ...string) *customer.Customer { + t.Helper() + cus, err := env.CustomerService.CreateCustomer(ctx, customer.CreateCustomerInput{ + Namespace: namespace, + CustomerMutate: customer.CustomerMutate{ + Key: lo.ToPtr(key), + Name: key, + UsageAttribution: &customer.CustomerUsageAttribution{SubjectKeys: subjectKeys}, + }, + }) + require.NoError(t, err) + return cus + } + + t.Run("ResolvesByKeyAndSubject", func(t *testing.T) { + // given: + // - customer A matched by its own key, customer B matched by one of its subject keys + // then: + // - each input key maps to the correct distinct customer + a := create(t, "a-key", "a-subject") + b := create(t, "b-key", "b-subject") + + got, err := env.CustomerService.GetCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ + Namespace: namespace, + Keys: []string{"a-key", "b-subject"}, + }) + require.NoError(t, err) + require.Len(t, got, 2) + assert.Equal(t, a.ID, got["a-key"].ID) + assert.Equal(t, b.ID, got["b-subject"].ID) + }) + + t.Run("KeyOverSubjectPrecedence", func(t *testing.T) { + // given: + // - "shared" is customer P's own key AND customer Q's subject key + // then: + // - the key owner (P) wins, resolved in the service + p := create(t, "shared", "p-subject") + _ = create(t, "q-key", "shared") + + got, err := env.CustomerService.GetCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ + Namespace: namespace, + Keys: []string{"shared"}, + }) + require.NoError(t, err) + require.Len(t, got, 1) + assert.Equal(t, p.ID, got["shared"].ID, "a direct customer-key match must win over a subject-key match") + }) + + t.Run("UnmatchedKeysAbsentNoError", func(t *testing.T) { + // given: + // - a mix of a known key and an unknown one + // then: + // - the unknown key is simply absent from the map (unlike the single-key method, the bulk + // method does NOT return a not-found error) + known := create(t, "known-key") - // given: - // - an empty key set - // when: - // - resolving customers by usage attribution - // then: - // - the service rejects it with a validation error before any DB query is built - _, err := env.CustomerService.GetCustomersByUsageAttribution(t.Context(), customer.GetCustomersByUsageAttributionInput{ - Namespace: namespace, - Keys: nil, + got, err := env.CustomerService.GetCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ + Namespace: namespace, + Keys: []string{"known-key", "totally-unknown"}, + }) + require.NoError(t, err) + require.Len(t, got, 1) + assert.Equal(t, known.ID, got["known-key"].ID) + _, ok := got["totally-unknown"] + assert.False(t, ok, "unmatched key must be absent from the result map") + }) + + t.Run("EmptyKeysFailsValidation", func(t *testing.T) { + // given: + // - an empty key set + // then: + // - the service rejects it with a validation error before any DB query is built + _, err := env.CustomerService.GetCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ + Namespace: namespace, + Keys: nil, + }) + require.Error(t, err, "empty key set must fail validation") + assert.True(t, models.IsGenericValidationError(err), "error must be a validation error") }) - require.Error(t, err, "empty key set must fail validation") - assert.True(t, models.IsGenericValidationError(err), "error must be a validation error") } // Test_GetCustomerByUsageAttribution covers the single-key wiring that Test_CustomerService's From 194b1e11ddf208d60be9c93588e1c10f2e93e43c Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Fri, 17 Jul 2026 16:41:08 +0200 Subject: [PATCH 10/13] test(customer): split usage-attribution benchmarks by layer The adapter benchmark inlined a copy of the service's key-over-subject precedence, mixing layers and risking drift from resolveCustomersByKeyWithPrecedence. Split into layer-honest benchmarks: - adapter package: BenchmarkCustomer(s)UsageAttribution*Query measure only the candidate query (cross_table_or vs union_all), no precedence. - service package: BenchmarkCustomer(s)UsageAttribution*Lookup drive the real service.Get(s)ByUsageAttribution end-to-end, exercising the real precedence. Shared seed/count/bulk-key fixtures move to a new leaf package customer/testutils/uabench (imports only the ent client, so both benchmark packages use it without an import cycle). The inline precedence copy is deleted; the seed SQL is unchanged. Run the timed loop of the bulk service benchmark inside b.Run so it gets a fresh, running timer: the previous flat form called b.StopTimer() during setup and b.ResetTimer() (which does not restart a stopped timer), so the loop was measured with the timer off and reported no ns/op. Test-only, no production change. --- ...stomer_usage_attribution_benchmark_test.go | 162 +++-------------- ...stomer_usage_attribution_benchmark_test.go | 165 ++++++++++++++++++ .../customer/testutils/uabench/uabench.go | 105 +++++++++++ 3 files changed, 293 insertions(+), 139 deletions(-) create mode 100644 openmeter/customer/service/customer_usage_attribution_benchmark_test.go create mode 100644 openmeter/customer/testutils/uabench/uabench.go diff --git a/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go b/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go index 05a394d743..0e932f86d0 100644 --- a/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go +++ b/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go @@ -3,37 +3,30 @@ package adapter import ( "context" "fmt" - "os" - "slices" - "strconv" "testing" "time" - entdb "github.com/openmeterio/openmeter/openmeter/ent/db" + "github.com/openmeterio/openmeter/openmeter/customer/testutils/uabench" customerdb "github.com/openmeterio/openmeter/openmeter/ent/db/customer" customersubjectsdb "github.com/openmeterio/openmeter/openmeter/ent/db/customersubjects" "github.com/openmeterio/openmeter/openmeter/testutils" ) -const ( - defaultUsageAttributionBenchmarkCustomers = 100_000 - usageAttributionBenchmarkSizeEnv = "CUSTOMER_USAGE_ATTRIBUTION_BENCH_CUSTOMERS" -) - -// BenchmarkCustomerUsageAttributionLookup compares the cross-table OR with the -// UNION ALL candidate-ID lookup. Seeding is outside the timed sections. Set -// CUSTOMER_USAGE_ATTRIBUTION_BENCH_CUSTOMERS to change the namespace size from -// the default 100,000 customers. -func BenchmarkCustomerUsageAttributionLookup(b *testing.B) { +// BenchmarkCustomerUsageAttributionQuery compares the cross-table OR with the UNION ALL candidate-ID +// query for a single key. It measures only the adapter-level candidate query (no precedence, no +// hydration); the full service path incl. key-over-subject precedence is benchmarked separately in +// the customer service package. Seeding is outside the timed sections. Set +// CUSTOMER_USAGE_ATTRIBUTION_BENCH_CUSTOMERS to change the namespace size from the default 100,000. +func BenchmarkCustomerUsageAttributionQuery(b *testing.B) { b.StopTimer() db := testutils.InitPostgresDB(b, testutils.PostgresDBStateEntMigrated) b.Cleanup(func() { db.Close(b) }) client := db.EntDriver.Client() - customerCount := usageAttributionBenchmarkCustomerCount(b) + customerCount := uabench.CustomerCount(b) namespace := "customer-usage-attribution-benchmark" - seedUsageAttributionBenchmark(b, client, namespace, customerCount) + uabench.Seed(b, client, namespace, customerCount) targetCustomerID := fmt.Sprintf("%026d", customerCount) now := time.Now().UTC() @@ -90,45 +83,6 @@ func BenchmarkCustomerUsageAttributionLookup(b *testing.B) { FirstID(ctx) }, }, - { - // method measures the FULL single-key path (merged predicate + WithSubjects - // hydration + domain mapping + Go-side precedence), not just the ID query, so it - // reflects the real customer/service.GetCustomerByUsageAttribution cost. The - // inline resolution mirrors service.resolveCustomersByKeyWithPrecedence over the raw candidate - // set (<=2 rows): a customer-key match wins over a subject-key match. - name: "method", - query: func(ctx context.Context) (string, error) { - q := client.Customer.Query().Where( - customerdb.Namespace(namespace), - customersMatchUsageAttributionKeys(namespace, []string{lookup.key}, now), - ) - q = WithSubjects(q, now) - entities, err := q.All(ctx) - if err != nil { - return "", err - } - var byKeyID, bySubjectID string - for _, entity := range entities { - c, err := CustomerFromDBEntity(*entity, nil) - if err != nil { - return "", err - } - if c.Key != nil && *c.Key == lookup.key { - byKeyID = c.ID - } - if c.UsageAttribution != nil && bySubjectID == "" && slices.Contains(c.UsageAttribution.SubjectKeys, lookup.key) { - bySubjectID = c.ID - } - } - if byKeyID != "" { - return byKeyID, nil - } - if bySubjectID != "" { - return bySubjectID, nil - } - return "", fmt.Errorf("no customer resolved for key %q", lookup.key) - }, - }, } for _, variant := range variants { @@ -161,15 +115,11 @@ func BenchmarkCustomerUsageAttributionLookup(b *testing.B) { } } -// bulkUsageAttributionBenchmarkKeyCount mirrors the governance QueryAccess OAS cap -// (@maxItems(100) on customer.keys), the primary bulk consumer's realistic upper bound. -const bulkUsageAttributionBenchmarkKeyCount = 100 - -// BenchmarkCustomersUsageAttributionBulkLookup compares the cross-table OR with the UNION ALL -// candidate-ID lookup for the bulk (multi-key) resolution path, mirroring -// BenchmarkCustomerUsageAttributionLookup for the single-key path. Seeding is outside the timed +// BenchmarkCustomersUsageAttributionBulkQuery compares the cross-table OR with the UNION ALL +// candidate-ID query for the bulk (multi-key) path, mirroring BenchmarkCustomerUsageAttributionQuery +// for the single-key path. Adapter-level query only (no precedence). Seeding is outside the timed // sections. Set CUSTOMER_USAGE_ATTRIBUTION_BENCH_CUSTOMERS to change the namespace size. -func BenchmarkCustomersUsageAttributionBulkLookup(b *testing.B) { +func BenchmarkCustomersUsageAttributionBulkQuery(b *testing.B) { b.StopTimer() db := testutils.InitPostgresDB(b) @@ -180,28 +130,17 @@ func BenchmarkCustomersUsageAttributionBulkLookup(b *testing.B) { b.Fatalf("create database schema: %v", err) } - customerCount := usageAttributionBenchmarkCustomerCount(b) - if customerCount < 2*bulkUsageAttributionBenchmarkKeyCount { + customerCount := uabench.CustomerCount(b) + if customerCount < 2*uabench.BulkKeyCount { b.Fatalf("%s must be at least %d to seed a non-overlapping bulk key set, got %d", - usageAttributionBenchmarkSizeEnv, 2*bulkUsageAttributionBenchmarkKeyCount, customerCount) + uabench.CustomersEnv, 2*uabench.BulkKeyCount, customerCount) } namespace := "customer-usage-attribution-bulk-benchmark" - seedUsageAttributionBenchmark(b, client, namespace, customerCount) + uabench.Seed(b, client, namespace, customerCount) now := time.Now().UTC() - - // Half the keys resolve via a direct customer-key match, half via a subject-key match on a - // disjoint customer range, mirroring a realistic mixed bulk lookup (governance resolves - // customer keys and subject keys through the same call). - half := bulkUsageAttributionBenchmarkKeyCount / 2 - keys := make([]string, 0, bulkUsageAttributionBenchmarkKeyCount) - for i := 1; i <= half; i++ { - keys = append(keys, fmt.Sprintf("customer-%d", i)) - } - for i := half + 1; i <= 2*half; i++ { - keys = append(keys, fmt.Sprintf("subject-%d", i)) - } + keys := uabench.BulkKeys(customerCount) variants := []struct { name string @@ -248,8 +187,8 @@ func BenchmarkCustomersUsageAttributionBulkLookup(b *testing.B) { if err != nil { b.Fatalf("warm up lookup: %v", err) } - if count != bulkUsageAttributionBenchmarkKeyCount { - b.Fatalf("warm up lookup returned %d customers, expected %d", count, bulkUsageAttributionBenchmarkKeyCount) + if count != uabench.BulkKeyCount { + b.Fatalf("warm up lookup returned %d customers, expected %d", count, uabench.BulkKeyCount) } b.ReportAllocs() @@ -262,66 +201,11 @@ func BenchmarkCustomersUsageAttributionBulkLookup(b *testing.B) { } b.StopTimer() b.ReportMetric(float64(customerCount), "customers") - b.ReportMetric(float64(bulkUsageAttributionBenchmarkKeyCount), "keys") + b.ReportMetric(float64(uabench.BulkKeyCount), "keys") - if count != bulkUsageAttributionBenchmarkKeyCount { - b.Fatalf("lookup returned %d customers, expected %d", count, bulkUsageAttributionBenchmarkKeyCount) + if count != uabench.BulkKeyCount { + b.Fatalf("lookup returned %d customers, expected %d", count, uabench.BulkKeyCount) } }) } } - -func usageAttributionBenchmarkCustomerCount(b *testing.B) int { - b.Helper() - - value := os.Getenv(usageAttributionBenchmarkSizeEnv) - if value == "" { - return defaultUsageAttributionBenchmarkCustomers - } - - count, err := strconv.Atoi(value) - if err != nil || count < 1 { - b.Fatalf("%s must be a positive integer, got %q", usageAttributionBenchmarkSizeEnv, value) - } - - return count -} - -func seedUsageAttributionBenchmark(b *testing.B, client *entdb.Client, namespace string, customerCount int) { - b.Helper() - - _, err := client.ExecContext(b.Context(), ` - INSERT INTO customers (id, namespace, created_at, updated_at, name, key) - SELECT - lpad(customer_number::text, 26, '0'), - $1, - now(), - now(), - 'customer-' || customer_number, - 'customer-' || customer_number - FROM generate_series(1, $2) AS customer_number - `, namespace, customerCount) - if err != nil { - b.Fatalf("seed customers: %v", err) - } - - _, err = client.ExecContext(b.Context(), ` - INSERT INTO customer_subjects (namespace, subject_key, created_at, customer_id) - SELECT - $1, - 'subject-' || customer_number, - now(), - lpad(customer_number::text, 26, '0') - FROM generate_series(1, $2) AS customer_number - `, namespace, customerCount) - if err != nil { - b.Fatalf("seed customer subjects: %v", err) - } - - if _, err = client.ExecContext(b.Context(), "ANALYZE customers"); err != nil { - b.Fatalf("analyze customers: %v", err) - } - if _, err = client.ExecContext(b.Context(), "ANALYZE customer_subjects"); err != nil { - b.Fatalf("analyze customer subjects: %v", err) - } -} diff --git a/openmeter/customer/service/customer_usage_attribution_benchmark_test.go b/openmeter/customer/service/customer_usage_attribution_benchmark_test.go new file mode 100644 index 0000000000..fc0c7bd1bb --- /dev/null +++ b/openmeter/customer/service/customer_usage_attribution_benchmark_test.go @@ -0,0 +1,165 @@ +package customerservice_test + +import ( + "context" + "fmt" + "testing" + + "github.com/openmeterio/openmeter/openmeter/customer" + customeradapter "github.com/openmeterio/openmeter/openmeter/customer/adapter" + customerservice "github.com/openmeterio/openmeter/openmeter/customer/service" + "github.com/openmeterio/openmeter/openmeter/customer/testutils/uabench" + "github.com/openmeterio/openmeter/openmeter/testutils" + noop "github.com/openmeterio/openmeter/openmeter/watermill/driver/noop" + "github.com/openmeterio/openmeter/openmeter/watermill/eventbus" +) + +// newUsageAttributionServiceBenchmark builds a real customer service over a freshly seeded namespace +// (100k customers by default) and returns it with the namespace and seeded size. It constructs the +// service directly (adapter + service, no-op event publisher) rather than via the *testing.T-only +// NewTestEnv, so it is usable from a *testing.B. +func newUsageAttributionServiceBenchmark(b *testing.B) (customer.Service, string, int) { + b.Helper() + + db := testutils.InitPostgresDB(b) + b.Cleanup(func() { db.Close(b) }) + + client := db.EntDriver.Client() + if err := client.Schema.Create(b.Context()); err != nil { + b.Fatalf("create database schema: %v", err) + } + + logger := testutils.NewDiscardLogger(b) + + publisher, err := eventbus.New(eventbus.Options{ + Publisher: &noop.Publisher{}, + TopicMapping: eventbus.TopicMapping{ + IngestEventsTopic: "bench-ingest-events", + SystemEventsTopic: "bench-system-events", + BalanceWorkerEventsTopic: "bench-balance-worker-events", + }, + Logger: logger, + }) + if err != nil { + b.Fatalf("build event publisher: %v", err) + } + + adapter, err := customeradapter.New(customeradapter.Config{Client: client, Logger: logger}) + if err != nil { + b.Fatalf("build customer adapter: %v", err) + } + + svc, err := customerservice.New(customerservice.Config{ + Adapter: adapter, + Publisher: publisher, + Logger: logger, + }) + if err != nil { + b.Fatalf("build customer service: %v", err) + } + + customerCount := uabench.CustomerCount(b) + namespace := "customer-usage-attribution-service-benchmark" + uabench.Seed(b, client, namespace, customerCount) + + return svc, namespace, customerCount +} + +// BenchmarkCustomerUsageAttributionLookup measures the full single-key service path — candidate +// query, subject hydration, and the real key-over-subject precedence in +// resolveCustomersByKeyWithPrecedence — via the public service method. The adapter-level query cost +// alone is benchmarked by BenchmarkCustomerUsageAttributionQuery in the adapter package. +func BenchmarkCustomerUsageAttributionLookup(b *testing.B) { + b.StopTimer() + + svc, namespace, customerCount := newUsageAttributionServiceBenchmark(b) + targetCustomerID := fmt.Sprintf("%026d", customerCount) + + lookups := []struct { + name string + key string + }{ + {name: "customer_key", key: fmt.Sprintf("customer-%d", customerCount)}, + {name: "subject_key", key: fmt.Sprintf("subject-%d", customerCount)}, + } + + for _, lookup := range lookups { + b.Run(lookup.name, func(b *testing.B) { + get := func(ctx context.Context) (string, error) { + c, err := svc.GetCustomerByUsageAttribution(ctx, customer.GetCustomerByUsageAttributionInput{ + Namespace: namespace, + Key: lookup.key, + }) + if err != nil { + return "", err + } + return c.ID, nil + } + + customerID, err := get(b.Context()) + if err != nil { + b.Fatalf("warm up lookup: %v", err) + } + if customerID != targetCustomerID { + b.Fatalf("warm up lookup returned customer %q, expected %q", customerID, targetCustomerID) + } + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := get(b.Context()); err != nil { + b.Fatalf("lookup customer: %v", err) + } + } + b.StopTimer() + b.ReportMetric(float64(customerCount), "customers") + }) + } +} + +// BenchmarkCustomersUsageAttributionBulkLookup measures the full bulk service path — one candidate +// query for the whole key set plus per-key precedence resolution into the returned map — via the +// public service method. +func BenchmarkCustomersUsageAttributionBulkLookup(b *testing.B) { + svc, namespace, customerCount := newUsageAttributionServiceBenchmark(b) + if customerCount < 2*uabench.BulkKeyCount { + b.Fatalf("%s must be at least %d to build a non-overlapping bulk key set, got %d", + uabench.CustomersEnv, 2*uabench.BulkKeyCount, customerCount) + } + + keys := uabench.BulkKeys(customerCount) + + get := func(ctx context.Context) (int, error) { + resolved, err := svc.GetCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ + Namespace: namespace, + Keys: keys, + }) + if err != nil { + return 0, err + } + return len(resolved), nil + } + + count, err := get(b.Context()) + if err != nil { + b.Fatalf("warm up lookup: %v", err) + } + if count != uabench.BulkKeyCount { + b.Fatalf("warm up lookup resolved %d customers, expected %d", count, uabench.BulkKeyCount) + } + + // The timed loop runs inside b.Run so it gets a fresh, running timer: setup above (which seeds + // 100k rows) is excluded from the measurement without relying on StopTimer/ResetTimer ordering. + b.Run(fmt.Sprintf("keys_%d", uabench.BulkKeyCount), func(b *testing.B) { + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := get(b.Context()); err != nil { + b.Fatalf("lookup customers: %v", err) + } + } + b.StopTimer() + b.ReportMetric(float64(customerCount), "customers") + b.ReportMetric(float64(uabench.BulkKeyCount), "keys") + }) +} diff --git a/openmeter/customer/testutils/uabench/uabench.go b/openmeter/customer/testutils/uabench/uabench.go new file mode 100644 index 0000000000..c551fcf052 --- /dev/null +++ b/openmeter/customer/testutils/uabench/uabench.go @@ -0,0 +1,105 @@ +// Package uabench holds shared fixtures for the customer usage-attribution benchmarks. It imports +// only the generated ent client (not the customer adapter or service packages), so both the +// adapter-layer benchmark (package adapter) and the service-layer benchmark can seed from one place +// without an import cycle. +package uabench + +import ( + "fmt" + "os" + "strconv" + "testing" + + entdb "github.com/openmeterio/openmeter/openmeter/ent/db" +) + +const ( + // CustomersEnv overrides the seeded namespace size. + CustomersEnv = "CUSTOMER_USAGE_ATTRIBUTION_BENCH_CUSTOMERS" + + defaultCustomers = 100_000 + + // BulkKeyCount mirrors the governance QueryAccess OAS cap (@maxItems(100) on customer.keys), + // the primary bulk consumer's realistic upper bound. + BulkKeyCount = 100 +) + +// CustomerCount returns the seeded namespace size, overridable via CustomersEnv (default 100,000). +func CustomerCount(tb testing.TB) int { + tb.Helper() + + value := os.Getenv(CustomersEnv) + if value == "" { + return defaultCustomers + } + + count, err := strconv.Atoi(value) + if err != nil || count < 1 { + tb.Fatalf("%s must be a positive integer, got %q", CustomersEnv, value) + } + + return count +} + +// BulkKeys builds a realistic mixed bulk key set of BulkKeyCount keys: half resolve via a direct +// customer-key match, half via a subject-key match on a disjoint customer range, mirroring how +// governance resolves customer keys and subject keys through the same call. Requires +// customerCount >= 2*BulkKeyCount. +func BulkKeys(customerCount int) []string { + half := BulkKeyCount / 2 + + keys := make([]string, 0, BulkKeyCount) + for i := 1; i <= half; i++ { + keys = append(keys, fmt.Sprintf("customer-%d", i)) + } + for i := half + 1; i <= 2*half; i++ { + keys = append(keys, fmt.Sprintf("subject-%d", i)) + } + + return keys +} + +// Seed bulk-inserts customerCount customers (key "customer-N") each with one subject ("subject-N") +// directly via SQL — orders of magnitude faster than the service create path — then ANALYZEs both +// tables so the planner has stats. Customer ids are the zero-padded ordinal, so callers can compute +// the target id for row N as lpad(N, 26, '0'). +func Seed(tb testing.TB, client *entdb.Client, namespace string, customerCount int) { + tb.Helper() + + ctx := tb.Context() + + _, err := client.ExecContext(ctx, ` + INSERT INTO customers (id, namespace, created_at, updated_at, name, key) + SELECT + lpad(customer_number::text, 26, '0'), + $1, + now(), + now(), + 'customer-' || customer_number, + 'customer-' || customer_number + FROM generate_series(1, $2) AS customer_number + `, namespace, customerCount) + if err != nil { + tb.Fatalf("seed customers: %v", err) + } + + _, err = client.ExecContext(ctx, ` + INSERT INTO customer_subjects (namespace, subject_key, created_at, customer_id) + SELECT + $1, + 'subject-' || customer_number, + now(), + lpad(customer_number::text, 26, '0') + FROM generate_series(1, $2) AS customer_number + `, namespace, customerCount) + if err != nil { + tb.Fatalf("seed customer subjects: %v", err) + } + + if _, err = client.ExecContext(ctx, "ANALYZE customers"); err != nil { + tb.Fatalf("analyze customers: %v", err) + } + if _, err = client.ExecContext(ctx, "ANALYZE customer_subjects"); err != nil { + tb.Fatalf("analyze customer subjects: %v", err) + } +} From 84e266c6070fca87c468106a2f0cac9c28f86766 Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Fri, 17 Jul 2026 16:59:55 +0200 Subject: [PATCH 11/13] test(customer): adapt usage-attribution tests to migrated test harness Rebase onto main changed the harness: NewTestEnv migrates internally and InitPostgresDB takes PostgresDBStateEntMigrated. Drop the leftover DBSchemaMigrate call and stale comment, and the now-redundant Schema.Create in the adapter and service benchmarks. --- .../adapter/customer_usage_attribution_benchmark_test.go | 5 +---- .../service/customer_usage_attribution_benchmark_test.go | 5 +---- openmeter/customer/service/service_test.go | 3 --- 3 files changed, 2 insertions(+), 11 deletions(-) diff --git a/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go b/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go index 0e932f86d0..7a10a20a9a 100644 --- a/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go +++ b/openmeter/customer/adapter/customer_usage_attribution_benchmark_test.go @@ -122,13 +122,10 @@ func BenchmarkCustomerUsageAttributionQuery(b *testing.B) { func BenchmarkCustomersUsageAttributionBulkQuery(b *testing.B) { b.StopTimer() - db := testutils.InitPostgresDB(b) + db := testutils.InitPostgresDB(b, testutils.PostgresDBStateEntMigrated) b.Cleanup(func() { db.Close(b) }) client := db.EntDriver.Client() - if err := client.Schema.Create(b.Context()); err != nil { - b.Fatalf("create database schema: %v", err) - } customerCount := uabench.CustomerCount(b) if customerCount < 2*uabench.BulkKeyCount { diff --git a/openmeter/customer/service/customer_usage_attribution_benchmark_test.go b/openmeter/customer/service/customer_usage_attribution_benchmark_test.go index fc0c7bd1bb..373dbc960d 100644 --- a/openmeter/customer/service/customer_usage_attribution_benchmark_test.go +++ b/openmeter/customer/service/customer_usage_attribution_benchmark_test.go @@ -21,13 +21,10 @@ import ( func newUsageAttributionServiceBenchmark(b *testing.B) (customer.Service, string, int) { b.Helper() - db := testutils.InitPostgresDB(b) + db := testutils.InitPostgresDB(b, testutils.PostgresDBStateEntMigrated) b.Cleanup(func() { db.Close(b) }) client := db.EntDriver.Client() - if err := client.Schema.Create(b.Context()); err != nil { - b.Fatalf("create database schema: %v", err) - } logger := testutils.NewDiscardLogger(b) diff --git a/openmeter/customer/service/service_test.go b/openmeter/customer/service/service_test.go index 7fea241699..953970c270 100644 --- a/openmeter/customer/service/service_test.go +++ b/openmeter/customer/service/service_test.go @@ -19,8 +19,6 @@ func Test_CustomerService(t *testing.T) { env.Close(t) }) - // Run database migrations - // Get new namespace ID namespace := customertestutils.NewTestNamespace(t) @@ -327,7 +325,6 @@ func Test_GetCustomerByUsageAttribution(t *testing.T) { t.Cleanup(func() { env.Close(t) }) - env.DBSchemaMigrate(t) namespace := customertestutils.NewTestNamespace(t) ctx := t.Context() From 7fff1473794ee781bcfec8a3053414f6a362c5b8 Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Fri, 17 Jul 2026 17:30:23 +0200 Subject: [PATCH 12/13] refactor(customer): drop unused logger from customer service The service Logger was added on this branch only for the ambiguous-match logging, which was removed; s.logger had no remaining uses. Remove the field from customerservice.Config and the Logger argument from all call sites. --- app/common/customer.go | 1 - .../service/customer_usage_attribution_benchmark_test.go | 1 - openmeter/customer/service/hooks/subjectcustomer_test.go | 1 - openmeter/customer/service/service.go | 8 -------- openmeter/customer/testutils/env.go | 1 - openmeter/entitlement/metered/lateevents_test.go | 1 - openmeter/entitlement/metered/utils_test.go | 1 - openmeter/entitlement/service/utils_test.go | 1 - openmeter/governance/service/service_test.go | 1 - openmeter/subject/service/hooks/customersubject_test.go | 1 - openmeter/subject/testutils/env.go | 1 - openmeter/subscription/testutils/customer.go | 1 - test/app/stripe/testenv.go | 1 - test/app/testenv.go | 1 - test/billing/suite.go | 1 - test/customer/testenv.go | 1 - test/entitlement/regression/framework_test.go | 1 - 17 files changed, 24 deletions(-) diff --git a/app/common/customer.go b/app/common/customer.go index b92bb80025..9f465e6caa 100644 --- a/app/common/customer.go +++ b/app/common/customer.go @@ -40,7 +40,6 @@ func NewCustomerService( service, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: eventPublisher, - Logger: logger.WithGroup("customer"), }) if err != nil { return nil, fmt.Errorf("failed to create customer service: %w", err) diff --git a/openmeter/customer/service/customer_usage_attribution_benchmark_test.go b/openmeter/customer/service/customer_usage_attribution_benchmark_test.go index 373dbc960d..66688348bb 100644 --- a/openmeter/customer/service/customer_usage_attribution_benchmark_test.go +++ b/openmeter/customer/service/customer_usage_attribution_benchmark_test.go @@ -49,7 +49,6 @@ func newUsageAttributionServiceBenchmark(b *testing.B) (customer.Service, string svc, err := customerservice.New(customerservice.Config{ Adapter: adapter, Publisher: publisher, - Logger: logger, }) if err != nil { b.Fatalf("build customer service: %v", err) diff --git a/openmeter/customer/service/hooks/subjectcustomer_test.go b/openmeter/customer/service/hooks/subjectcustomer_test.go index 9df6c26f44..5e3dfd71f4 100644 --- a/openmeter/customer/service/hooks/subjectcustomer_test.go +++ b/openmeter/customer/service/hooks/subjectcustomer_test.go @@ -337,7 +337,6 @@ func NewTestEnv(t *testing.T) *TestEnv { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, - Logger: logger, }) require.NoErrorf(t, err, "initializing subject service must not fail") require.NotNilf(t, subjectAdapter, "subject service must not be nil") diff --git a/openmeter/customer/service/service.go b/openmeter/customer/service/service.go index b7288e0cc0..1ebdcc0396 100644 --- a/openmeter/customer/service/service.go +++ b/openmeter/customer/service/service.go @@ -2,7 +2,6 @@ package customerservice import ( "errors" - "log/slog" "github.com/openmeterio/openmeter/openmeter/customer" "github.com/openmeterio/openmeter/openmeter/watermill/eventbus" @@ -15,7 +14,6 @@ type Service struct { adapter customer.Adapter requestValidatorRegistry customer.RequestValidatorRegistry publisher eventbus.Publisher - logger *slog.Logger hooks models.ServiceHookRegistry[customer.Customer] } @@ -27,7 +25,6 @@ func (s *Service) RegisterHooks(hooks ...models.ServiceHook[customer.Customer]) type Config struct { Adapter customer.Adapter Publisher eventbus.Publisher - Logger *slog.Logger } func (c Config) Validate() error { @@ -39,10 +36,6 @@ func (c Config) Validate() error { return errors.New("publisher cannot be null") } - if c.Logger == nil { - return errors.New("logger cannot be null") - } - return nil } @@ -55,7 +48,6 @@ func New(config Config) (*Service, error) { adapter: config.Adapter, requestValidatorRegistry: customer.NewRequestValidatorRegistry(), publisher: config.Publisher, - logger: config.Logger, hooks: models.ServiceHookRegistry[customer.Customer]{}, }, nil } diff --git a/openmeter/customer/testutils/env.go b/openmeter/customer/testutils/env.go index 04e25ca44e..3314df6cd5 100644 --- a/openmeter/customer/testutils/env.go +++ b/openmeter/customer/testutils/env.go @@ -105,7 +105,6 @@ func NewTestEnv(t *testing.T) *TestEnv { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, - Logger: logger, }) require.NoErrorf(t, err, "initializing subject service must not fail") require.NotNilf(t, customerService, "subject service must not be nil") diff --git a/openmeter/entitlement/metered/lateevents_test.go b/openmeter/entitlement/metered/lateevents_test.go index a835d1564a..0ad2d9b962 100644 --- a/openmeter/entitlement/metered/lateevents_test.go +++ b/openmeter/entitlement/metered/lateevents_test.go @@ -158,7 +158,6 @@ func TestGetEntitlementBalanceConsistency(t *testing.T) { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: mockPublisher, - Logger: testLogger, }) require.NoError(t, err) diff --git a/openmeter/entitlement/metered/utils_test.go b/openmeter/entitlement/metered/utils_test.go index 04a4dad2d1..1d31c80770 100644 --- a/openmeter/entitlement/metered/utils_test.go +++ b/openmeter/entitlement/metered/utils_test.go @@ -139,7 +139,6 @@ func setupConnector(t *testing.T) (meteredentitlement.Connector, *dependencies) customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: mockPublisher, - Logger: testLogger, }) require.NoError(t, err) diff --git a/openmeter/entitlement/service/utils_test.go b/openmeter/entitlement/service/utils_test.go index cea0df2b52..26fe81dc81 100644 --- a/openmeter/entitlement/service/utils_test.go +++ b/openmeter/entitlement/service/utils_test.go @@ -134,7 +134,6 @@ func setupDependecies(t *testing.T) (entitlement.Service, *dependencies) { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: eventbus.NewMock(t), - Logger: testLogger, }) if err != nil { t.Fatalf("failed to create customer service: %v", err) diff --git a/openmeter/governance/service/service_test.go b/openmeter/governance/service/service_test.go index 2c46ae5c19..81352f6f3d 100644 --- a/openmeter/governance/service/service_test.go +++ b/openmeter/governance/service/service_test.go @@ -91,7 +91,6 @@ func setupTestDeps(t *testing.T) *testDeps { customerSvc, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: eventbus.NewMock(t), - Logger: logger, }) require.NoError(t, err) diff --git a/openmeter/subject/service/hooks/customersubject_test.go b/openmeter/subject/service/hooks/customersubject_test.go index 0e910ee037..c1051a08bb 100644 --- a/openmeter/subject/service/hooks/customersubject_test.go +++ b/openmeter/subject/service/hooks/customersubject_test.go @@ -289,7 +289,6 @@ func NewTestEnv(t *testing.T) *TestEnv { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, - Logger: logger, }) require.NoErrorf(t, err, "initializing subject service must not fail") require.NotNilf(t, subjectAdapter, "subject service must not be nil") diff --git a/openmeter/subject/testutils/env.go b/openmeter/subject/testutils/env.go index 9d95fbfcc0..0c78dd3892 100644 --- a/openmeter/subject/testutils/env.go +++ b/openmeter/subject/testutils/env.go @@ -119,7 +119,6 @@ func NewTestEnv(t *testing.T) *TestEnv { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, - Logger: logger, }) require.NoErrorf(t, err, "initializing subject service must not fail") require.NotNilf(t, customerService, "subject service must not be nil") diff --git a/openmeter/subscription/testutils/customer.go b/openmeter/subscription/testutils/customer.go index a491ddb798..c2c1c244f7 100644 --- a/openmeter/subscription/testutils/customer.go +++ b/openmeter/subscription/testutils/customer.go @@ -63,7 +63,6 @@ func NewCustomerService(t *testing.T, dbDeps *DBDeps) customer.Service { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: eventbus.NewMock(t), - Logger: testutils.NewLogger(t), }) if err != nil { t.Fatalf("failed to create customer service: %v", err) diff --git a/test/app/stripe/testenv.go b/test/app/stripe/testenv.go index 426e8d8af7..a46bfa4293 100644 --- a/test/app/stripe/testenv.go +++ b/test/app/stripe/testenv.go @@ -114,7 +114,6 @@ func NewTestEnv(t *testing.T, ctx context.Context) (TestEnv, error) { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, - Logger: logger, }) if err != nil { return nil, fmt.Errorf("failed to create customer service: %w", err) diff --git a/test/app/testenv.go b/test/app/testenv.go index 9266b1f329..8f50fb17a1 100644 --- a/test/app/testenv.go +++ b/test/app/testenv.go @@ -96,7 +96,6 @@ func NewTestEnv(t *testing.T, ctx context.Context) (TestEnv, error) { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, - Logger: logger, }) if err != nil { return nil, fmt.Errorf("failed to create customer service: %w", err) diff --git a/test/billing/suite.go b/test/billing/suite.go index cc2548d28c..60cdf2aff1 100644 --- a/test/billing/suite.go +++ b/test/billing/suite.go @@ -164,7 +164,6 @@ func (s *BaseSuite) setupSuite() { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, - Logger: slog.Default(), }) require.NoError(t, err) s.CustomerService = customerService diff --git a/test/customer/testenv.go b/test/customer/testenv.go index 0df41e8847..c422fd4fe3 100644 --- a/test/customer/testenv.go +++ b/test/customer/testenv.go @@ -187,7 +187,6 @@ func NewTestEnv(t *testing.T, ctx context.Context) (TestEnv, error) { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: publisher, - Logger: logger, }) if err != nil { return nil, err diff --git a/test/entitlement/regression/framework_test.go b/test/entitlement/regression/framework_test.go index e07c857d6a..3641fe2304 100644 --- a/test/entitlement/regression/framework_test.go +++ b/test/entitlement/regression/framework_test.go @@ -145,7 +145,6 @@ func setupDependencies(t *testing.T) Dependencies { customerService, err := customerservice.New(customerservice.Config{ Adapter: customerAdapter, Publisher: mockPublisher, - Logger: log, }) require.NoError(t, err) From 2468628fb37e62a6bcfe96ce94a98ed7eafeb3a9 Mon Sep 17 00:00:00 2001 From: Gergely Tamas Kurucz Date: Tue, 21 Jul 2026 16:51:55 +0200 Subject: [PATCH 13/13] refactor(customer): return map[string]*Customer from GetCustomersByUsageAttribution Align the bulk resolver's semantics with the single-key lookup and the repo's featureresolver.BatchResolve convention: every input key is present in the returned map, with a nil value marking not-found (was: value map, absent = not-found). Callers can distinguish requested-but-missing from not-requested, and the map holds pointers instead of copied Customer structs. --- openmeter/customer/service.go | 2 +- openmeter/customer/service/customer.go | 30 +++++++++++-------- openmeter/customer/service/customer_test.go | 18 +++++------ ...stomer_usage_attribution_benchmark_test.go | 10 ++++++- openmeter/customer/service/service_test.go | 16 +++++----- openmeter/governance/service/service.go | 6 ++-- openmeter/server/server_test.go | 2 +- 7 files changed, 50 insertions(+), 34 deletions(-) diff --git a/openmeter/customer/service.go b/openmeter/customer/service.go index 6d2aede5d0..63991071b1 100644 --- a/openmeter/customer/service.go +++ b/openmeter/customer/service.go @@ -26,6 +26,6 @@ type CustomerService interface { DeleteCustomer(ctx context.Context, customer DeleteCustomerInput) error GetCustomer(ctx context.Context, customer GetCustomerInput) (*Customer, error) GetCustomerByUsageAttribution(ctx context.Context, input GetCustomerByUsageAttributionInput) (*Customer, error) - GetCustomersByUsageAttribution(ctx context.Context, input GetCustomersByUsageAttributionInput) (map[string]Customer, error) + GetCustomersByUsageAttribution(ctx context.Context, input GetCustomersByUsageAttributionInput) (map[string]*Customer, error) UpdateCustomer(ctx context.Context, params UpdateCustomerInput) (*Customer, error) } diff --git a/openmeter/customer/service/customer.go b/openmeter/customer/service/customer.go index c5c30a6f3c..cf400f15bb 100644 --- a/openmeter/customer/service/customer.go +++ b/openmeter/customer/service/customer.go @@ -150,19 +150,20 @@ func (s *Service) GetCustomerByUsageAttribution(ctx context.Context, input custo return nil, err } - c, ok := resolved[input.Key] - if !ok { + c := resolved[input.Key] + if c == nil { return nil, models.NewGenericNotFoundError( fmt.Errorf("customer with subject key %s not found in %s namespace", input.Key, input.Namespace), ) } - return &c, nil + return c, nil } // GetCustomersByUsageAttribution resolves multiple customers by usage attribution keys in a single // query, mapping each input key to the customer it matches with key-over-subject precedence applied. -func (s *Service) GetCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) (map[string]customer.Customer, error) { +// Every input key is present in the returned map; keys that match no customer map to a nil value. +func (s *Service) GetCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) (map[string]*customer.Customer, error) { if err := input.Validate(); err != nil { return nil, models.NewGenericValidationError( fmt.Errorf("error getting customers by usage attribution: %w", err), @@ -175,7 +176,7 @@ func (s *Service) GetCustomersByUsageAttribution(ctx context.Context, input cust // resolveCustomersByUsageAttribution runs the shared usage-attribution lookup: fetch the raw // candidate customers from the adapter, then apply key-over-subject precedence. Both the single-key // and bulk service methods route through this so their core resolution stays identical. -func (s *Service) resolveCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) (map[string]customer.Customer, error) { +func (s *Service) resolveCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) (map[string]*customer.Customer, error) { customers, err := s.adapter.GetCustomersByUsageAttribution(ctx, input) if err != nil { return nil, err @@ -188,13 +189,15 @@ func (s *Service) resolveCustomersByUsageAttribution(ctx context.Context, input // either by the customer's own key or by one of its subject keys; when a key matches both a // distinct key-owner and a distinct subject-owner, the key-owner takes precedence. Key-over-subject // collisions are structurally rare and resolved deterministically here, so they are not surfaced as -// errors or logs; investigate the underlying data via the database if ever needed. Keys with no -// match are absent from the returned map. -func resolveCustomersByKeyWithPrecedence(customers []customer.Customer, keys []string) map[string]customer.Customer { - byKey := make(map[string]customer.Customer, len(customers)) - bySubject := make(map[string]customer.Customer, len(customers)) +// errors or logs; investigate the underlying data via the database if ever needed. Every input key +// is present in the returned map; keys that match no customer have a nil value. +func resolveCustomersByKeyWithPrecedence(customers []customer.Customer, keys []string) map[string]*customer.Customer { + byKey := make(map[string]*customer.Customer, len(customers)) + bySubject := make(map[string]*customer.Customer, len(customers)) + + for i := range customers { + c := &customers[i] - for _, c := range customers { if c.Key != nil { byKey[*c.Key] = c } @@ -208,7 +211,7 @@ func resolveCustomersByKeyWithPrecedence(customers []customer.Customer, keys []s } } - resolved := make(map[string]customer.Customer, len(keys)) + resolved := make(map[string]*customer.Customer, len(keys)) for _, k := range keys { if keyOwner, ok := byKey[k]; ok { @@ -218,7 +221,10 @@ func resolveCustomersByKeyWithPrecedence(customers []customer.Customer, keys []s if subjectOwner, ok := bySubject[k]; ok { resolved[k] = subjectOwner + continue } + + resolved[k] = nil // present, but no matching customer } return resolved diff --git a/openmeter/customer/service/customer_test.go b/openmeter/customer/service/customer_test.go index 8c6a3952ba..a2b6703120 100644 --- a/openmeter/customer/service/customer_test.go +++ b/openmeter/customer/service/customer_test.go @@ -22,19 +22,19 @@ func Test_resolveCustomersByKeyWithPrecedence(t *testing.T) { t.Run("MatchByKey", func(t *testing.T) { resolved := resolveCustomersByKeyWithPrecedence([]customer.Customer{customerA}, []string{"key-a"}) - assert.Equal(t, map[string]customer.Customer{"key-a": customerA}, resolved) + assert.Equal(t, map[string]*customer.Customer{"key-a": &customerA}, resolved) }) t.Run("MatchBySubject", func(t *testing.T) { resolved := resolveCustomersByKeyWithPrecedence([]customer.Customer{customerA}, []string{"subject-a"}) - assert.Equal(t, map[string]customer.Customer{"subject-a": customerA}, resolved) + assert.Equal(t, map[string]*customer.Customer{"subject-a": &customerA}, resolved) }) - t.Run("UnmatchedKeyIsAbsent", func(t *testing.T) { + t.Run("UnmatchedKeyIsNil", func(t *testing.T) { resolved := resolveCustomersByKeyWithPrecedence([]customer.Customer{customerA}, []string{"no-such-key"}) - assert.Empty(t, resolved) + assert.Equal(t, map[string]*customer.Customer{"no-such-key": nil}, resolved) }) t.Run("KeyOwnerTakesPrecedenceOverDistinctSubjectOwner", func(t *testing.T) { @@ -55,7 +55,7 @@ func Test_resolveCustomersByKeyWithPrecedence(t *testing.T) { []string{"shared"}, ) - assert.Equal(t, map[string]customer.Customer{"shared": sharedKeyCustomer}, resolved) + assert.Equal(t, map[string]*customer.Customer{"shared": &sharedKeyCustomer}, resolved) }) t.Run("SameCustomerMatchedByOwnKeyAndSubjectKey", func(t *testing.T) { @@ -69,7 +69,7 @@ func Test_resolveCustomersByKeyWithPrecedence(t *testing.T) { resolved := resolveCustomersByKeyWithPrecedence([]customer.Customer{selfMatched}, []string{"dual"}) - assert.Equal(t, map[string]customer.Customer{"dual": selfMatched}, resolved) + assert.Equal(t, map[string]*customer.Customer{"dual": &selfMatched}, resolved) }) t.Run("CrossCollisionResolvesEachKeyToItsOwnKeyOwner", func(t *testing.T) { @@ -96,9 +96,9 @@ func Test_resolveCustomersByKeyWithPrecedence(t *testing.T) { []string{"key-1", "key-2"}, ) - assert.Equal(t, map[string]customer.Customer{ - "key-1": crossA, - "key-2": crossB, + assert.Equal(t, map[string]*customer.Customer{ + "key-1": &crossA, + "key-2": &crossB, }, resolved) }) } diff --git a/openmeter/customer/service/customer_usage_attribution_benchmark_test.go b/openmeter/customer/service/customer_usage_attribution_benchmark_test.go index 66688348bb..25d17e6bfd 100644 --- a/openmeter/customer/service/customer_usage_attribution_benchmark_test.go +++ b/openmeter/customer/service/customer_usage_attribution_benchmark_test.go @@ -125,6 +125,8 @@ func BenchmarkCustomersUsageAttributionBulkLookup(b *testing.B) { keys := uabench.BulkKeys(customerCount) + // Count non-nil entries: every input key is present in the result map, with a nil value for + // keys that resolved to no customer. get := func(ctx context.Context) (int, error) { resolved, err := svc.GetCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ Namespace: namespace, @@ -133,7 +135,13 @@ func BenchmarkCustomersUsageAttributionBulkLookup(b *testing.B) { if err != nil { return 0, err } - return len(resolved), nil + matched := 0 + for _, c := range resolved { + if c != nil { + matched++ + } + } + return matched, nil } count, err := get(b.Context()) diff --git a/openmeter/customer/service/service_test.go b/openmeter/customer/service/service_test.go index 953970c270..e6ce871541 100644 --- a/openmeter/customer/service/service_test.go +++ b/openmeter/customer/service/service_test.go @@ -220,7 +220,7 @@ func Test_CustomerService(t *testing.T) { // Test_GetCustomersByUsageAttribution protects the bulk method's contract with the same rigor as // Test_GetCustomerByUsageAttribution does the single-key one: each input key maps to its customer -// with key-over-subject precedence, and unmatched keys are simply absent (no error). The predicate +// with key-over-subject precedence, and unmatched keys are present with a nil value (no error). The predicate // itself is covered by the adapter's TestGetCustomersByUsageAttribution, and the fully symmetric // {K1,K2} cross-collision by Test_resolveCustomersByKeyWithPrecedence — that state cannot be built // through the service here, because CreateCustomer rejects a customer whose key overlaps another @@ -283,12 +283,12 @@ func Test_GetCustomersByUsageAttribution(t *testing.T) { assert.Equal(t, p.ID, got["shared"].ID, "a direct customer-key match must win over a subject-key match") }) - t.Run("UnmatchedKeysAbsentNoError", func(t *testing.T) { + t.Run("UnmatchedKeysNil", func(t *testing.T) { // given: // - a mix of a known key and an unknown one // then: - // - the unknown key is simply absent from the map (unlike the single-key method, the bulk - // method does NOT return a not-found error) + // - every input key is present in the map; the unknown key has a nil value (unlike the + // single-key method, the bulk method does NOT return a not-found error) known := create(t, "known-key") got, err := env.CustomerService.GetCustomersByUsageAttribution(ctx, customer.GetCustomersByUsageAttributionInput{ @@ -296,10 +296,12 @@ func Test_GetCustomersByUsageAttribution(t *testing.T) { Keys: []string{"known-key", "totally-unknown"}, }) require.NoError(t, err) - require.Len(t, got, 1) + require.Len(t, got, 2) + require.NotNil(t, got["known-key"]) assert.Equal(t, known.ID, got["known-key"].ID) - _, ok := got["totally-unknown"] - assert.False(t, ok, "unmatched key must be absent from the result map") + unknown, ok := got["totally-unknown"] + assert.True(t, ok, "every input key must be present in the result map") + assert.Nil(t, unknown, "unmatched key must have a nil value") }) t.Run("EmptyKeysFailsValidation", func(t *testing.T) { diff --git a/openmeter/governance/service/service.go b/openmeter/governance/service/service.go index 74fe9ebb34..ca5c7d5501 100644 --- a/openmeter/governance/service/service.go +++ b/openmeter/governance/service/service.go @@ -303,9 +303,9 @@ func (s *service) resolveCustomers(ctx context.Context, input governance.QueryAc // Iterate the input keys in order so matched keys and not-found errors keep input ordering. for _, key := range input.CustomerKeys { - cus, ok := keyToCustomer[key] + cus := keyToCustomer[key] - if !ok { + if cus == nil { queryErrors = append(queryErrors, governance.QueryError{ CustomerKey: key, Code: governance.QueryErrorCustomerNotFound, @@ -318,7 +318,7 @@ func (s *service) resolveCustomers(ctx context.Context, input governance.QueryAc rc.matched = append(rc.matched, key) } else { customerMap[cus.ID] = &resolvedCustomer{ - customer: cus, + customer: *cus, matched: []string{key}, } } diff --git a/openmeter/server/server_test.go b/openmeter/server/server_test.go index 67a9112b76..77ea96527e 100644 --- a/openmeter/server/server_test.go +++ b/openmeter/server/server_test.go @@ -1458,7 +1458,7 @@ func (n NoopCustomerService) GetCustomerByUsageAttribution(ctx context.Context, return &customer.Customer{}, nil } -func (n NoopCustomerService) GetCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) (map[string]customer.Customer, error) { +func (n NoopCustomerService) GetCustomersByUsageAttribution(ctx context.Context, input customer.GetCustomersByUsageAttributionInput) (map[string]*customer.Customer, error) { return nil, nil }