diff --git a/account/account.go b/account/account.go index 87597e5feda..86e5a5f22ff 100644 --- a/account/account.go +++ b/account/account.go @@ -15,6 +15,15 @@ import ( "github.com/prebid/prebid-server/v4/util/jsonutil" ) +// TypedAccountFetcher is an optional interface that a fetcher may implement to +// return a fully-derived, immutable *config.Account directly, skipping the +// per-request JSON unmarshal + defaults-merge + derive done on the legacy path. +// The Fetchers 2.0 (cachekit) account fetcher implements this; when present, +// GetAccount uses it. Legacy fetchers do not, and take the JSON path unchanged. +type TypedAccountFetcher interface { + FetchAccountTyped(ctx context.Context, accountID string) (*config.Account, []error) +} + // GetAccount looks up the config.Account object referenced by the given accountID, with access rules applied func GetAccount(ctx context.Context, cfg *config.Configuration, fetcher stored_requests.AccountFetcher, accountID string, me metrics.MetricsEngine) (account *config.Account, errs []error) { if cfg.AccountRequired && accountID == metrics.PublisherUnknown { @@ -23,6 +32,16 @@ func GetAccount(ctx context.Context, cfg *config.Configuration, fetcher stored_r }} } + if typed, ok := fetcher.(TypedAccountFetcher); ok { + return getAccountTyped(ctx, cfg, typed, accountID) + } + return getAccountJSON(ctx, cfg, fetcher, accountID) +} + +// getAccountJSON is the legacy account resolution path: it fetches raw +// (defaults-merged) JSON, unmarshals it, unpacks DSA defaults and computes the +// derived config on every call. +func getAccountJSON(ctx context.Context, cfg *config.Configuration, fetcher stored_requests.AccountFetcher, accountID string) (account *config.Account, errs []error) { if accountJSON, accErrs := fetcher.FetchAccount(ctx, cfg.AccountDefaultsJSON(), accountID); len(accErrs) > 0 || accountJSON == nil { // accountID does not reference a valid account for _, e := range accErrs { @@ -70,6 +89,59 @@ func GetAccount(ctx context.Context, cfg *config.Configuration, fetcher stored_r return nil, errs } + applyIPMaskingDefaults(account) + return account, nil +} + +// getAccountTyped is the Fetchers 2.0 account resolution path. The fetcher returns +// a fully-derived, immutable *config.Account (unmarshal + DSA + derive + IP masking +// were done once, at cache insert). This path only applies the not-found fallback +// and the per-request access gating. +func getAccountTyped(ctx context.Context, cfg *config.Configuration, fetcher TypedAccountFetcher, accountID string) (account *config.Account, errs []error) { + fetched, accErrs := fetcher.FetchAccountTyped(ctx, accountID) + if len(accErrs) > 0 { + // A malformed account is a hard error, mirroring the legacy path where the + // unmarshal/DSA failure returns immediately rather than falling back to defaults. + for _, e := range accErrs { + if _, ok := e.(*errortypes.MalformedAcct); ok { + return nil, accErrs + } + } + // Otherwise (not-found, or a swallowed backend error) fall through to the + // AccountDefaults fallback, matching the legacy not-found branch. + fetched = nil + } + if fetched == nil { + if cfg.AccountRequired && cfg.AccountDefaults.Disabled { + return nil, []error{&errortypes.AcctRequired{ + Message: "Prebid-server could not verify the Account ID. Please reach out to the prebid server host.", + }} + } + // Make a copy of AccountDefaults instead of taking a reference, + // to preserve original accountID in case is needed to check NonStandardPublisherMap + pubAccount := cfg.AccountDefaults + pubAccount.ID = accountID + account = &pubAccount + } else { + // Fully-derived, immutable account returned straight from the cache. + account = fetched + } + if account.Disabled { + errs = append(errs, &errortypes.AccountDisabled{ + Message: fmt.Sprintf("Prebid-server has disabled Account ID: %s, please reach out to the prebid server host.", accountID), + }) + return nil, errs + } + + // No-op for the cached (already-masked) account; corrects the defaults copy. + applyIPMaskingDefaults(account) + return account, nil +} + +// applyIPMaskingDefaults falls back to the default IPv4/IPv6 masking bit sizes when +// the configured values are invalid. Re-running it on an already-valid account is a +// read-only no-op. +func applyIPMaskingDefaults(account *config.Account) { if ipV6Err := account.Privacy.IPv6Config.Validate(nil); len(ipV6Err) > 0 { account.Privacy.IPv6Config.AnonKeepBits = iputil.IPv6DefaultMaskingBitSize } @@ -77,8 +149,6 @@ func GetAccount(ctx context.Context, cfg *config.Configuration, fetcher stored_r if ipV4Err := account.Privacy.IPv4Config.Validate(nil); len(ipV4Err) > 0 { account.Privacy.IPv4Config.AnonKeepBits = iputil.IPv4DefaultMaskingBitSize } - - return account, nil } // TCF2Enforcements maps enforcement algo string values to their integer representation and is diff --git a/account/cachekit.go b/account/cachekit.go new file mode 100644 index 00000000000..3a0adf772d1 --- /dev/null +++ b/account/cachekit.go @@ -0,0 +1,244 @@ +package account + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/benbjohnson/clock" + jsonpatch "gopkg.in/evanphx/json-patch.v5" + + "github.com/prebid/prebid-server/v4/cachekit" + "github.com/prebid/prebid-server/v4/config" + "github.com/prebid/prebid-server/v4/errortypes" + "github.com/prebid/prebid-server/v4/logger" + "github.com/prebid/prebid-server/v4/metrics" + "github.com/prebid/prebid-server/v4/stored_requests" + "github.com/prebid/prebid-server/v4/util/jsonutil" +) + +// cacheKitSubsystem is the metrics subsystem label for the account cache. +const cacheKitSubsystem = "account" + +// CacheKitAccountFetcher is the Fetchers 2.0 account fetcher. It embeds the +// underlying source fetcher (file / db / http / multi) so it continues to satisfy +// stored_requests.AllFetcher, and adds FetchAccountTyped which serves fully-derived, +// immutable *config.Account values from a cachekit engine. +type CacheKitAccountFetcher struct { + stored_requests.AllFetcher + engine *cachekit.Fetcher[string, *config.Account] +} + +// FetchAccountTyped implements account.TypedAccountFetcher. +func (f *CacheKitAccountFetcher) FetchAccountTyped(ctx context.Context, accountID string) (*config.Account, []error) { + account, err := f.engine.Get(ctx, accountID) + if err != nil { + if errors.Is(err, cachekit.ErrNotFound) { + return nil, []error{stored_requests.NotFoundError{ID: accountID, DataType: "Account"}} + } + return nil, []error{err} + } + return account, nil +} + +// NewCacheKitAccountFetcher wires a cachekit engine in front of an existing account +// source. The sources emit raw, unmerged account rows and the shared transform applies +// the account defaults once (at cache insert); this fetcher adds the typed cache, +// single-flight coalescing and optional negative caching. clk may be nil (a real clock +// is used). metricsEngine may be nil (metrics are not recorded). +func NewCacheKitAccountFetcher(source stored_requests.AllFetcher, cfg config.CacheKitConfig, defaults json.RawMessage, clk clock.Clock, metricsEngine metrics.MetricsEngine) (*CacheKitAccountFetcher, error) { + var cache cachekit.Cache[string, *config.Account] + switch cfg.Type { + case "none": + cache = cachekit.NoCache[string, *config.Account]{} + case "", "lru": + lru, err := cachekit.NewLRUCache[string, *config.Account](cfg.MaxEntries, clk) + if err != nil { + return nil, err + } + cache = lru + default: + return nil, fmt.Errorf("accounts.cache.type %q is not supported (expected none or lru)", cfg.Type) + } + + var negatives *cachekit.NegativeStore[string] + if cfg.Negative.Enabled { + n, err := cachekit.NewNegativeStore[string](cfg.Negative.MaxEntries, cfg.Negative.TTL(), clk) + if err != nil { + // Negative caching is an optimization, not a correctness requirement. If it + // can't be built (e.g. a bad accounts.cache.negative.max_entries), warn and + // continue without it rather than aborting account fetcher startup; not-found + // lookups will just fall through to the backend each time. + logger.Warnf("account cachekit: negative caching disabled, failed to initialize: %v", err) + } else { + negatives = n + } + } + + var recorder cachekit.Recorder + if metricsEngine != nil { + recorder = metricsRecorder{engine: metricsEngine, subsystem: cacheKitSubsystem} + } + + // Freshness (refresh) axis: ttl (serve-stale + background revalidation), none + // (never revalidate / load-once), or preload (bulk warm at startup then ttl). + effectiveTTL := cfg.TTL() + serveStale := cfg.ServeStale + var preload cachekit.BulkSource[string] + switch cfg.Refresh { + case "", config.RefreshTTL: + serveStale = true + case config.RefreshNone: + effectiveTTL = 0 // never revalidate + case config.RefreshPreload: + serveStale = true + bulk, ok := source.(stored_requests.AllAccountsFetcher) + if !ok { + return nil, fmt.Errorf("accounts.cache.refresh %q requires an account source that supports bulk loading (FetchAllAccounts)", cfg.Refresh) + } + preload = accountBulkSource{fetcher: bulk} + // NOTE: "delta-poll" (event-driven Save/Invalidation, mirroring v1's http_events / + // cache-events producers) is intentionally not implemented: no known deployment + // pushes live account updates, so it would be untested, unused code. It can be added + // later as a background mechanism without touching Source/Transform/Cache. + default: + return nil, fmt.Errorf("accounts.cache.refresh %q is not supported (expected ttl, none or preload)", cfg.Refresh) + } + + engine := cachekit.New(cachekit.Params[string, *config.Account]{ + Source: accountSource{fetcher: source}, + Transform: newAccountTransform(defaults), + Cache: cache, + TTL: effectiveTTL, + Negatives: negatives, + Coalesce: cfg.CoalesceRequests, + ServeStale: serveStale, + RevalidateTimeout: cfg.RevalidateTimeout(), + Preload: preload, + Clock: clk, + Metrics: recorder, + }) + engine.Start(context.Background()) + + return &CacheKitAccountFetcher{AllFetcher: source, engine: engine}, nil +} + +// metricsRecorder adapts a metrics.MetricsEngine to the cachekit.Recorder interface, +// emitting the dedicated cachekit_* metrics under the given subsystem label. +type metricsRecorder struct { + engine metrics.MetricsEngine + subsystem string +} + +func (r metricsRecorder) CacheHit() { + r.engine.RecordCacheKitResult(r.subsystem, metrics.CacheKitResultHit) +} +func (r metricsRecorder) CacheMiss() { + r.engine.RecordCacheKitResult(r.subsystem, metrics.CacheKitResultMiss) +} +func (r metricsRecorder) CacheNegative() { + r.engine.RecordCacheKitResult(r.subsystem, metrics.CacheKitResultNegative) +} + +func (r metricsRecorder) BackendFetch(result string, d time.Duration) { + var mapped metrics.CacheKitBackendResult + switch result { + case "ok": + mapped = metrics.CacheKitBackendOK + case "notfound": + mapped = metrics.CacheKitBackendNotFound + default: + mapped = metrics.CacheKitBackendError + } + r.engine.RecordCacheKitBackendFetch(r.subsystem, mapped, d) +} + +// accountSource adapts an existing stored_requests account fetcher into a +// cachekit.Source. It requests the raw, unmerged account row (defaults are applied +// once, downstream, by the shared transform) and reuses the backend's not-found +// classification: a NotFoundError becomes an absent map key (cachekit's "not found" +// convention); any other error is a systemic failure and is not cached. +type accountSource struct { + fetcher stored_requests.AccountFetcher +} + +func (s accountSource) Fetch(ctx context.Context, keys []string) (map[string]json.RawMessage, error) { + out := make(map[string]json.RawMessage, len(keys)) + for _, id := range keys { + // nil defaults => the backend returns the raw row without merging. The shared + // transform applies defaults, so the single-key and bulk paths merge in one place. + raw, errs := s.fetcher.FetchAccount(ctx, nil, id) + if len(errs) > 0 { + if isNotFoundErr(errs) { + continue // absent key => definitive not-found for this id + } + return nil, errors.Join(errs...) + } + out[id] = raw + } + return out, nil +} + +func isNotFoundErr(errs []error) bool { + for _, e := range errs { + if _, ok := e.(stored_requests.NotFoundError); ok { + return true + } + } + return false +} + +// accountBulkSource adapts a stored_requests.AllAccountsFetcher into a cachekit.BulkSource. +// It returns the raw, unmerged account rows as-is; the shared transform applies defaults +// downstream, so this path and the single-key path merge in exactly one place. +type accountBulkSource struct { + fetcher stored_requests.AllAccountsFetcher +} + +func (s accountBulkSource) FetchAll(ctx context.Context) (map[string]json.RawMessage, error) { + data, errs := s.fetcher.FetchAllAccounts(ctx) + if len(errs) > 0 { + return nil, errors.Join(errs...) + } + return data, nil +} + +// newAccountTransform returns the single normalization step for accounts. It merges the +// account defaults into the raw row, then unmarshals, unpacks DSA defaults, fills the ID, +// and computes the derived + IP-masking config. It runs once per id at cache insert, so +// the per-request read path does no merge, unmarshal, DSA unpack, derive or IP masking. +// Both the single-key and bulk sources feed it raw, unmerged rows, so the defaults-merge +// lives here in exactly one place. +func newAccountTransform(defaults json.RawMessage) cachekit.TransformFunc[string, *config.Account] { + return func(accountID string, raw json.RawMessage) (*config.Account, error) { + merged := raw + if defaults != nil { + m, err := jsonpatch.MergePatch(defaults, raw) + if err != nil { + return nil, &errortypes.MalformedAcct{ + Message: fmt.Sprintf("The prebid-server account config for account id \"%s\" is malformed. Please reach out to the prebid server host.", accountID), + } + } + merged = m + } + account := &config.Account{} + if err := jsonutil.UnmarshalValid(merged, account); err != nil { + return nil, &errortypes.MalformedAcct{ + Message: fmt.Sprintf("The prebid-server account config for account id \"%s\" is malformed. Please reach out to the prebid server host.", accountID), + } + } + if err := config.UnpackDSADefault(account.Privacy.DSA); err != nil { + return nil, &errortypes.MalformedAcct{ + Message: fmt.Sprintf("The prebid-server account config DSA for account id \"%s\" is malformed. Please reach out to the prebid server host.", accountID), + } + } + if len(account.ID) == 0 { + account.ID = accountID + } + setDerivedConfig(account) + applyIPMaskingDefaults(account) + return account, nil + } +} diff --git a/account/cachekit_test.go b/account/cachekit_test.go new file mode 100644 index 00000000000..2ce9997da84 --- /dev/null +++ b/account/cachekit_test.go @@ -0,0 +1,277 @@ +package account + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/benbjohnson/clock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/prebid/prebid-server/v4/config" + "github.com/prebid/prebid-server/v4/errortypes" + metricsconfig "github.com/prebid/prebid-server/v4/metrics/config" + "github.com/prebid/prebid-server/v4/openrtb_ext" + "github.com/prebid/prebid-server/v4/stored_requests" + "github.com/prebid/prebid-server/v4/stored_requests/caches/memory" +) + +// mockAllFetcher is a minimal stored_requests.AllFetcher for the v2 account tests. +// Only FetchAccount is exercised; the other methods satisfy the interface. +type mockAllFetcher struct { + accounts map[string]json.RawMessage + accountCalls int + bulkCalls int +} + +func (m *mockAllFetcher) FetchRequests(_ context.Context, _ []string, _ []string) (map[string]json.RawMessage, map[string]json.RawMessage, []error) { + return nil, nil, nil +} + +func (m *mockAllFetcher) FetchResponses(_ context.Context, _ []string) (map[string]json.RawMessage, []error) { + return nil, nil +} + +func (m *mockAllFetcher) FetchCategories(_ context.Context, _, _, _ string) (string, error) { + return "", nil +} + +func (m *mockAllFetcher) FetchAccount(_ context.Context, _ json.RawMessage, accountID string) (json.RawMessage, []error) { + m.accountCalls++ + raw, ok := m.accounts[accountID] + if !ok { + return nil, []error{stored_requests.NotFoundError{ID: accountID, DataType: "Account"}} + } + return raw, nil +} + +// FetchAllAccounts makes the mock a bulk-capable source for refresh: preload. +func (m *mockAllFetcher) FetchAllAccounts(_ context.Context) (map[string]json.RawMessage, []error) { + m.bulkCalls++ + return m.accounts, nil +} + +func newV2Fetcher(t *testing.T, fetcher stored_requests.AllFetcher) *CacheKitAccountFetcher { + t.Helper() + cfg := config.CacheKitConfig{Type: "lru", MaxEntries: 100, TTLSeconds: 3600} + v2, err := NewCacheKitAccountFetcher(fetcher, cfg, json.RawMessage(`{}`), clock.NewMock(), nil) + require.NoError(t, err) + return v2 +} + +func TestV2GetAccountTypedHit(t *testing.T) { + fetcher := &mockAllFetcher{accounts: map[string]json.RawMessage{ + "pub-1": json.RawMessage(`{"id":"pub-1"}`), + }} + v2 := newV2Fetcher(t, fetcher) + cfg := &config.Configuration{} + + account, errs := GetAccount(context.Background(), cfg, v2, "pub-1", nil) + require.Empty(t, errs) + require.NotNil(t, account) + assert.Equal(t, "pub-1", account.ID) + // Derived config is computed once at cache insert. + assert.NotNil(t, account.GDPR.PurposeConfigs, "derived config should be populated") + + // Second lookup is served from the typed cache without hitting the source. + account, errs = GetAccount(context.Background(), cfg, v2, "pub-1", nil) + require.Empty(t, errs) + assert.Equal(t, "pub-1", account.ID) + assert.Equal(t, 1, fetcher.accountCalls, "second GetAccount should be a cache hit") +} + +func TestV2GetAccountAppliesDefaultsAndDerivedConfigOnce(t *testing.T) { + defaults := json.RawMessage(`{ + "gdpr": { + "basic_enforcement_vendors": ["appnexus"], + "purpose1": { + "enforce_algo": "basic", + "vendor_exceptions": ["rubicon"] + }, + "special_feature1": { + "vendor_exceptions": ["appnexus"] + } + }, + "privacy": { + "dsa": { + "default": "{\"dsarequired\":1,\"pubrender\":2,\"transparency\":[{\"domain\":\"test.com\"}]}" + } + } + }`) + fetcher := &mockAllFetcher{accounts: map[string]json.RawMessage{ + "pub-1": json.RawMessage(`{"id":"pub-1"}`), + }} + v2, err := NewCacheKitAccountFetcher(fetcher, config.CacheKitConfig{ + Type: "lru", + MaxEntries: 100, + TTLSeconds: 3600, + }, defaults, clock.NewMock(), nil) + require.NoError(t, err) + + account, errs := GetAccount(context.Background(), &config.Configuration{}, v2, "pub-1", nil) + require.Empty(t, errs) + require.NotNil(t, account) + + assert.Contains(t, account.GDPR.BasicEnforcementVendorsMap, "appnexus") + assert.Contains(t, account.GDPR.Purpose1.VendorExceptionMap, "rubicon") + assert.Contains(t, account.GDPR.SpecialFeature1.VendorExceptionMap, openrtb_ext.BidderName("appnexus")) + assert.Equal(t, config.TCF2BasicEnforcement, account.GDPR.Purpose1.EnforceAlgoID) + require.NotNil(t, account.Privacy.DSA) + require.NotNil(t, account.Privacy.DSA.DefaultUnpacked) + assert.Equal(t, int8(1), *account.Privacy.DSA.DefaultUnpacked.Required) + assert.Equal(t, int8(2), *account.Privacy.DSA.DefaultUnpacked.PubRender) + assert.Equal(t, "test.com", account.Privacy.DSA.DefaultUnpacked.Transparency[0].Domain) + + // A second lookup is a typed-cache hit: the source is not called again, and + // defaults/DSA/derived map work is not repeated through the fetcher path. + account, errs = GetAccount(context.Background(), &config.Configuration{}, v2, "pub-1", nil) + require.Empty(t, errs) + require.NotNil(t, account) + assert.Equal(t, 1, fetcher.accountCalls) +} + +func TestV2WrappingLegacyCacheCanMaskBackendChangesAfterTTL(t *testing.T) { + clk := clock.NewMock() + source := &mockAllFetcher{accounts: map[string]json.RawMessage{ + "pub-1": json.RawMessage(`{"id":"pub-1","disabled":false}`), + }} + legacyCachedFetcher := stored_requests.WithCache(source, stored_requests.Cache{ + Accounts: memory.NewCache(0, 0, "Accounts"), + }, &metricsconfig.NilMetricsEngine{}) + v2, err := NewCacheKitAccountFetcher(legacyCachedFetcher, config.CacheKitConfig{ + Type: "lru", + MaxEntries: 100, + TTLSeconds: 1, + }, json.RawMessage(`{}`), clk, nil) + require.NoError(t, err) + + account, errs := GetAccount(context.Background(), &config.Configuration{}, v2, "pub-1", nil) + require.Empty(t, errs) + require.NotNil(t, account) + assert.False(t, account.Disabled) + assert.Equal(t, 1, source.accountCalls) + + // The real source changes, and v2 TTL expires. Because v2 is wrapped around + // the legacy byte cache, the reload is satisfied by that old cache instead of + // calling the real source again. + source.accounts["pub-1"] = json.RawMessage(`{"id":"pub-1","disabled":true}`) + clk.Add(2 * time.Second) + + account, errs = GetAccount(context.Background(), &config.Configuration{}, v2, "pub-1", nil) + require.Empty(t, errs) + require.NotNil(t, account) + assert.False(t, account.Disabled, "v2 reloaded from the legacy cache, not the changed source") + assert.Equal(t, 1, source.accountCalls, "backend source is hidden behind the legacy v1 cache") +} + +func TestV2GetAccountNotFoundFallsBackToDefaults(t *testing.T) { + fetcher := &mockAllFetcher{accounts: map[string]json.RawMessage{}} + v2 := newV2Fetcher(t, fetcher) + cfg := &config.Configuration{} + + account, errs := GetAccount(context.Background(), cfg, v2, "unknown", nil) + require.Empty(t, errs) + require.NotNil(t, account) + assert.Equal(t, "unknown", account.ID, "not-found should fall back to AccountDefaults with the requested ID") +} + +func TestV2GetAccountMalformedReturnsError(t *testing.T) { + fetcher := &mockAllFetcher{accounts: map[string]json.RawMessage{ + "bad": json.RawMessage(`{`), + }} + v2 := newV2Fetcher(t, fetcher) + cfg := &config.Configuration{} + + account, errs := GetAccount(context.Background(), cfg, v2, "bad", nil) + require.Nil(t, account) + require.NotEmpty(t, errs) + _, isMalformed := errs[0].(*errortypes.MalformedAcct) + assert.True(t, isMalformed, "malformed account JSON should surface a MalformedAcct error") +} + +func TestV2GetAccountDisabled(t *testing.T) { + fetcher := &mockAllFetcher{accounts: map[string]json.RawMessage{ + "off": json.RawMessage(`{"id":"off","disabled":true}`), + }} + v2 := newV2Fetcher(t, fetcher) + cfg := &config.Configuration{} + + account, errs := GetAccount(context.Background(), cfg, v2, "off", nil) + require.Nil(t, account) + require.NotEmpty(t, errs) + _, isDisabled := errs[0].(*errortypes.AccountDisabled) + assert.True(t, isDisabled, "disabled account should surface an AccountDisabled error") +} + +func TestV2RefreshPreloadWarmsCache(t *testing.T) { + fetcher := &mockAllFetcher{accounts: map[string]json.RawMessage{ + "pub-1": json.RawMessage(`{"id":"pub-1"}`), + }} + cfg := config.CacheKitConfig{Type: "lru", MaxEntries: 100, TTLSeconds: 3600, Refresh: "preload"} + v2, err := NewCacheKitAccountFetcher(fetcher, cfg, json.RawMessage(`{}`), clock.NewMock(), nil) + require.NoError(t, err) + assert.Equal(t, 1, fetcher.bulkCalls, "preload should perform a single bulk fetch at startup") + + account, errs := GetAccount(context.Background(), &config.Configuration{}, v2, "pub-1", nil) + require.Empty(t, errs) + assert.Equal(t, "pub-1", account.ID) + assert.Equal(t, 0, fetcher.accountCalls, "preloaded account should be served without a per-key fetch") +} + +func TestV2RefreshTTLServesStaleByDefault(t *testing.T) { + clk := clock.NewMock() + fetcher := &mockAllFetcher{accounts: map[string]json.RawMessage{ + "pub-1": json.RawMessage(`{"id":"pub-1","disabled":false}`), + }} + cfg := config.CacheKitConfig{Type: "lru", MaxEntries: 100, TTLSeconds: 1, Refresh: "ttl"} + v2, err := NewCacheKitAccountFetcher(fetcher, cfg, json.RawMessage(`{}`), clk, nil) + require.NoError(t, err) + + account, errs := GetAccount(context.Background(), &config.Configuration{}, v2, "pub-1", nil) + require.Empty(t, errs) + require.NotNil(t, account) + assert.False(t, account.Disabled) + assert.Equal(t, 1, fetcher.accountCalls) + + fetcher.accounts["pub-1"] = json.RawMessage(`{"id":"pub-1","disabled":true}`) + clk.Add(2 * time.Second) + + account, errs = GetAccount(context.Background(), &config.Configuration{}, v2, "pub-1", nil) + require.Empty(t, errs) + require.NotNil(t, account) + assert.False(t, account.Disabled, "ttl mode should return stale data immediately and refresh in the background") + assert.Eventually(t, func() bool { return fetcher.accountCalls == 2 }, time.Second, 5*time.Millisecond) +} + +func TestV2RefreshPreloadUnsupportedSourceErrors(t *testing.T) { + // A source that does not implement AllAccountsFetcher cannot preload. + var plain stored_requests.AllFetcher = notBulkFetcher{} + cfg := config.CacheKitConfig{Type: "lru", MaxEntries: 100, TTLSeconds: 3600, Refresh: "preload"} + _, err := NewCacheKitAccountFetcher(plain, cfg, json.RawMessage(`{}`), clock.NewMock(), nil) + require.Error(t, err) +} + +func TestV2RefreshUnknownModeErrors(t *testing.T) { + fetcher := &mockAllFetcher{accounts: map[string]json.RawMessage{}} + cfg := config.CacheKitConfig{Type: "lru", MaxEntries: 100, TTLSeconds: 3600, Refresh: "bogus"} + _, err := NewCacheKitAccountFetcher(fetcher, cfg, json.RawMessage(`{}`), clock.NewMock(), nil) + require.Error(t, err) +} + +// notBulkFetcher is an AllFetcher that does NOT implement AllAccountsFetcher. +type notBulkFetcher struct{} + +func (notBulkFetcher) FetchRequests(_ context.Context, _ []string, _ []string) (map[string]json.RawMessage, map[string]json.RawMessage, []error) { + return nil, nil, nil +} +func (notBulkFetcher) FetchResponses(_ context.Context, _ []string) (map[string]json.RawMessage, []error) { + return nil, nil +} +func (notBulkFetcher) FetchCategories(_ context.Context, _, _, _ string) (string, error) { + return "", nil +} +func (notBulkFetcher) FetchAccount(_ context.Context, _ json.RawMessage, accountID string) (json.RawMessage, []error) { + return nil, []error{stored_requests.NotFoundError{ID: accountID, DataType: "Account"}} +} diff --git a/cachekit/cache.go b/cachekit/cache.go new file mode 100644 index 00000000000..cc504be963c --- /dev/null +++ b/cachekit/cache.go @@ -0,0 +1,74 @@ +package cachekit + +import ( + "time" + + "github.com/benbjohnson/clock" + lru "github.com/hashicorp/golang-lru/v2" +) + +// entry is the positive-store value: the composed typed value plus the time after +// which it is considered stale and should be refreshed in the background. A zero +// refreshAfter means the entry never goes stale (load-once / mirror modes). +type entry[V any] struct { + v V + refreshAfter time.Time +} + +// LRUCache is a bounded, serve-stale read-through cache backed by +// hashicorp/golang-lru/v2. Entries are never evicted by time: past refreshAfter +// they are still returned (flagged stale) so the read path never blocks on the +// backend, and the engine refreshes them in the background. LRU capacity is the +// only eviction. It is safe for concurrent use. +type LRUCache[K comparable, V any] struct { + lru *lru.Cache[K, entry[V]] + clock clock.Clock +} + +// NewLRUCache builds an LRU cache holding up to maxEntries values. +func NewLRUCache[K comparable, V any](maxEntries int, clk clock.Clock) (*LRUCache[K, V], error) { + if clk == nil { + clk = clock.New() + } + l, err := lru.New[K, entry[V]](maxEntries) + if err != nil { + return nil, err + } + return &LRUCache[K, V]{lru: l, clock: clk}, nil +} + +// Get returns the value if present, and whether it is stale (past its refresh +// time). Stale entries are still returned; the caller decides whether to trigger a +// background refresh. +func (c *LRUCache[K, V]) Get(key K) (V, bool, bool) { + e, ok := c.lru.Get(key) + if !ok { + var zero V + return zero, false, false + } + stale := !e.refreshAfter.IsZero() && c.clock.Now().After(e.refreshAfter) + return e.v, true, stale +} + +// Save stores v under key. ttl <= 0 means the entry never goes stale. +func (c *LRUCache[K, V]) Save(key K, v V, ttl time.Duration) { + var refreshAfter time.Time + if ttl > 0 { + refreshAfter = c.clock.Now().Add(ttl) + } + c.lru.Add(key, entry[V]{v: v, refreshAfter: refreshAfter}) +} + +// Invalidate removes key from the cache if present. +func (c *LRUCache[K, V]) Invalidate(key K) { + c.lru.Remove(key) +} + +// NoCache is a pass-through cache: every Get is a miss and Save is a no-op. Paired +// with the engine's single-flight coalescing it yields "always fetch, still +// deduplicate" behaviour for direct-source / live tenants. +type NoCache[K comparable, V any] struct{} + +func (NoCache[K, V]) Get(K) (V, bool, bool) { var zero V; return zero, false, false } +func (NoCache[K, V]) Save(K, V, time.Duration) {} +func (NoCache[K, V]) Invalidate(K) {} diff --git a/cachekit/cache_test.go b/cachekit/cache_test.go new file mode 100644 index 00000000000..b5f4c905a5b --- /dev/null +++ b/cachekit/cache_test.go @@ -0,0 +1,103 @@ +package cachekit + +import ( + "testing" + "time" + + "github.com/benbjohnson/clock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestLRUCacheSaveGetAndStaleState(t *testing.T) { + clk := clock.NewMock() + cache, err := NewLRUCache[string, string](2, clk) + require.NoError(t, err) + + cache.Save("a", "v1", time.Hour) + + v, ok, stale := cache.Get("a") + assert.True(t, ok) + assert.False(t, stale) + assert.Equal(t, "v1", v) + + clk.Add(2 * time.Hour) + + v, ok, stale = cache.Get("a") + assert.True(t, ok, "stale entries are still returned; the fetcher decides how to refresh") + assert.True(t, stale) + assert.Equal(t, "v1", v) +} + +func TestLRUCacheNonPositiveTTLNeverStales(t *testing.T) { + clk := clock.NewMock() + cache, err := NewLRUCache[string, string](2, clk) + require.NoError(t, err) + + cache.Save("a", "v1", 0) + clk.Add(24 * time.Hour) + + v, ok, stale := cache.Get("a") + assert.True(t, ok) + assert.False(t, stale) + assert.Equal(t, "v1", v) +} + +func TestLRUCacheEvictsLeastRecentlyUsedEntry(t *testing.T) { + cache, err := NewLRUCache[string, string](2, clock.NewMock()) + require.NoError(t, err) + + cache.Save("a", "v1", time.Hour) + cache.Save("b", "v2", time.Hour) + _, ok, _ := cache.Get("a") + require.True(t, ok, "touch a so b becomes least recently used") + cache.Save("c", "v3", time.Hour) + + _, ok, _ = cache.Get("b") + assert.False(t, ok, "least recently used entry should be evicted") + + v, ok, stale := cache.Get("a") + assert.True(t, ok) + assert.False(t, stale) + assert.Equal(t, "v1", v) + + v, ok, stale = cache.Get("c") + assert.True(t, ok) + assert.False(t, stale) + assert.Equal(t, "v3", v) +} + +func TestLRUCacheInvalidateRemovesEntry(t *testing.T) { + cache, err := NewLRUCache[string, string](2, clock.NewMock()) + require.NoError(t, err) + + cache.Save("a", "v1", time.Hour) + cache.Invalidate("a") + + _, ok, stale := cache.Get("a") + assert.False(t, ok) + assert.False(t, stale) +} + +func TestNewLRUCacheRejectsInvalidSize(t *testing.T) { + cache, err := NewLRUCache[string, string](0, clock.NewMock()) + + assert.Nil(t, cache) + assert.Error(t, err) +} + +func TestNoCacheAlwaysMisses(t *testing.T) { + cache := NoCache[string, string]{} + + cache.Save("a", "v1", time.Hour) + v, ok, stale := cache.Get("a") + assert.Empty(t, v) + assert.False(t, ok) + assert.False(t, stale) + + cache.Invalidate("a") + v, ok, stale = cache.Get("a") + assert.Empty(t, v) + assert.False(t, ok) + assert.False(t, stale) +} diff --git a/cachekit/cachekit.go b/cachekit/cachekit.go new file mode 100644 index 00000000000..b037d7a8536 --- /dev/null +++ b/cachekit/cachekit.go @@ -0,0 +1,213 @@ +// Package cachekit is a small, generic read-through fetching engine shared by +// Prebid Server subsystems (accounts today; GVL / stored data / currency later). +// +// It is intentionally higher-level than any single subsystem: a subsystem picks +// a Source (where raw bytes come from), a Transform (raw bytes -> typed value), +// and a Cache (retention policy), and cachekit wires them together. When serve-stale +// is enabled, stale entries are served immediately and revalidated in the background; +// optional single-flight coalescing collapses concurrent misses for the same key +// into one upstream call per pod. +// +// The cache stores the composed typed value V, not raw JSON. Transform runs once +// per key at insert time; a cache hit is a pure lookup with no unmarshal. +// +// The package is split by concern: +// - cachekit.go — the engine (Params, Fetcher, Get, load, preload). +// - contracts.go — the interfaces a consumer implements (Source, Cache, ...). +// - revalidate.go — the background serve-stale revalidation mechanism. +// - cache.go — the LRU / no-op positive cache implementations. +// - negative.go — the negative (definitive-verdict) store. +package cachekit + +import ( + "context" + "fmt" + "time" + + "github.com/benbjohnson/clock" + "golang.org/x/sync/singleflight" +) + +// Params configures a Fetcher. Source, Transform and Cache are required. +type Params[K comparable, V any] struct { + Source Source[K] + Transform TransformFunc[K, V] + Cache Cache[K, V] + TTL time.Duration + Negatives *NegativeStore[K] // nil disables negative caching + Coalesce bool // opt-in single-flight coalescing of concurrent misses (default off) + ServeStale bool // opt-in: past TTL, serve the stale value and revalidate in the background (default off = expire + synchronous reload) + RevalidateTimeout time.Duration // maximum duration for a background stale revalidation; <= 0 uses a safe default + Preload BulkSource[K] // if set, the whole corpus is fetched once at Start to warm the cache + Clock clock.Clock // nil defaults to a real clock + Metrics Recorder // nil defaults to a no-op recorder +} + +// Fetcher is the generic read-through engine. Construct it with New. +type Fetcher[K comparable, V any] struct { + source Source[K] + transform TransformFunc[K, V] + cache Cache[K, V] + ttl time.Duration + negatives *NegativeStore[K] + coalesce bool + serveStale bool + preload BulkSource[K] + clock clock.Clock + metrics Recorder + group singleflight.Group + reval *revalidator[K] + revalTimeout time.Duration +} + +const defaultRevalidateTimeout = 10 * time.Second + +// New builds a Fetcher from the given params. +func New[K comparable, V any](p Params[K, V]) *Fetcher[K, V] { + if p.Clock == nil { + p.Clock = clock.New() + } + if p.Metrics == nil { + p.Metrics = noopRecorder{} + } + if p.RevalidateTimeout <= 0 { + p.RevalidateTimeout = defaultRevalidateTimeout + } + return &Fetcher[K, V]{ + source: p.Source, + transform: p.Transform, + cache: p.Cache, + ttl: p.TTL, + negatives: p.Negatives, + coalesce: p.Coalesce, + serveStale: p.ServeStale, + preload: p.Preload, + clock: p.Clock, + metrics: p.Metrics, + reval: newRevalidator[K](p.Clock, revalidateBackoff), + revalTimeout: p.RevalidateTimeout, + } +} + +// Start warms the cache when a Preload source is configured: it fetches the whole +// corpus once and seeds it. It is a no-op otherwise. Preload is best-effort — if the +// bulk fetch fails, the cache is left cold and fills lazily via the read path. +// Callers should invoke it once after construction. +func (f *Fetcher[K, V]) Start(ctx context.Context) { + if f.preload == nil { + return + } + start := f.clock.Now() + raw, err := f.preload.FetchAll(ctx) + if err != nil { + f.metrics.BackendFetch("error", f.clock.Now().Sub(start)) + return + } + f.metrics.BackendFetch("ok", f.clock.Now().Sub(start)) + for key, bytes := range raw { + v, err := f.transform(key, bytes) + if err != nil { + continue // skip malformed entries; they surface on demand + } + f.cache.Save(key, v, f.ttl) + } +} + +// Close is a no-op; background revalidations are fire-and-forget goroutines. +func (f *Fetcher[K, V]) Close() {} + +// Get returns the typed value for key. On a fresh cache hit it is a pure lookup +// with no upstream call and no unmarshal. Past TTL the behaviour depends on the +// ServeStale option: when enabled, the stale value is returned immediately and +// revalidated in the background (the read path never blocks on the backend); when +// disabled (default), a stale entry is treated as expired and reloaded +// synchronously. On a miss it fetches from the source; when coalescing is enabled, +// concurrent callers for the same key collapse into a single upstream fetch. It +// returns ErrNotFound when the key does not exist, and re-serves a cached verdict +// error (not-found or malformed) when negative caching is enabled. +func (f *Fetcher[K, V]) Get(ctx context.Context, key K) (V, error) { + var zero V + + if v, ok, stale := f.cache.Get(key); ok { + if !stale || f.serveStale { + f.metrics.CacheHit() + if stale { + f.triggerRevalidate(key) + } + return v, nil + } + // Stale with serve-stale disabled: reload synchronously (classic TTL expiry). + f.metrics.CacheMiss() + return f.fetch(ctx, key) + } + if f.negatives != nil { + if verr, ok := f.negatives.isCached(key); ok { + f.metrics.CacheNegative() + return zero, verr + } + } + f.metrics.CacheMiss() + return f.fetch(ctx, key) +} + +// fetch loads the key from the source, applying single-flight coalescing when it is +// enabled so concurrent callers collapse into one upstream call. +func (f *Fetcher[K, V]) fetch(ctx context.Context, key K) (V, error) { + if !f.coalesce { + return f.load(ctx, key) + } + var zero V + res, err, _ := f.group.Do(fmt.Sprint(key), func() (interface{}, error) { + // Another goroutine may have filled the cache (fresh) while we waited. + if v, ok, stale := f.cache.Get(key); ok && !stale { + return v, nil + } + return f.load(ctx, key) + }) + if err != nil { + return zero, err + } + return res.(V), nil +} + +// load performs a blocking upstream fetch, classification and cache insert for a +// cold key. Transient failures are never cached; definitive verdicts (not-found, +// malformed) are negative-cached with the real error when negative caching is on. +func (f *Fetcher[K, V]) load(ctx context.Context, key K) (V, error) { + var zero V + + start := f.clock.Now() + found, err := f.source.Fetch(ctx, []K{key}) + dur := f.clock.Now().Sub(start) + + if err != nil { + // Systemic/transient failure: never cache, never negative-cache. + f.metrics.BackendFetch("error", dur) + return zero, err + } + raw, ok := found[key] + if !ok { + // Definitive per-key not-found. + f.metrics.BackendFetch("notfound", dur) + if f.negatives != nil { + f.negatives.mark(key, ErrNotFound) + } + return zero, ErrNotFound + } + + v, err := f.transform(key, raw) + if err != nil { + // Malformed value: a permanent verdict. Surface the error and, when negative + // caching is on, remember it (error-preserving) so we re-serve the same + // malformed error without re-hitting the backend for a short window. + f.metrics.BackendFetch("error", dur) + if f.negatives != nil { + f.negatives.mark(key, err) + } + return zero, err + } + + f.cache.Save(key, v, f.ttl) + f.metrics.BackendFetch("ok", dur) + return v, nil +} diff --git a/cachekit/cachekit_test.go b/cachekit/cachekit_test.go new file mode 100644 index 00000000000..aa51eeed271 --- /dev/null +++ b/cachekit/cachekit_test.go @@ -0,0 +1,448 @@ +package cachekit + +import ( + "context" + "encoding/json" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/benbjohnson/clock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// stubSource is a test Source. It counts calls, can block (to exercise +// coalescing), and returns a fixed error when set. +type stubSource struct { + calls int32 + data map[string]json.RawMessage + err error + block chan struct{} +} + +func (s *stubSource) Fetch(_ context.Context, keys []string) (map[string]json.RawMessage, error) { + atomic.AddInt32(&s.calls, 1) + if s.block != nil { + <-s.block + } + if s.err != nil { + return nil, s.err + } + out := make(map[string]json.RawMessage, len(keys)) + for _, k := range keys { + if v, ok := s.data[k]; ok { + out[k] = v + } + } + return out, nil +} + +func (s *stubSource) callCount() int { return int(atomic.LoadInt32(&s.calls)) } + +type timeoutOnceSource struct { + calls int32 + data map[string]json.RawMessage +} + +func (s *timeoutOnceSource) Fetch(ctx context.Context, keys []string) (map[string]json.RawMessage, error) { + call := atomic.AddInt32(&s.calls, 1) + if call == 2 { + <-ctx.Done() + return nil, ctx.Err() + } + out := make(map[string]json.RawMessage, len(keys)) + for _, k := range keys { + if v, ok := s.data[k]; ok { + out[k] = v + } + } + return out, nil +} + +func (s *timeoutOnceSource) callCount() int { return int(atomic.LoadInt32(&s.calls)) } + +func identityTransform(_ string, raw json.RawMessage) (string, error) { + return string(raw), nil +} + +// bulkStub is a test BulkSource. +type bulkStub struct { + data map[string]json.RawMessage + err error +} + +func (b bulkStub) FetchAll(_ context.Context) (map[string]json.RawMessage, error) { + return b.data, b.err +} + +func TestPreloadSeedsCache(t *testing.T) { + clk := clock.NewMock() + cache, err := NewLRUCache[string, string](100, clk) + require.NoError(t, err) + readSrc := &stubSource{data: map[string]json.RawMessage{}} // read path source is empty + f := New(Params[string, string]{ + Source: readSrc, + Transform: identityTransform, + Cache: cache, + TTL: time.Hour, + Preload: bulkStub{data: map[string]json.RawMessage{"a": json.RawMessage(`v1`)}}, + Clock: clk, + }) + + f.Start(context.Background()) + + v, err := f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, "v1", v) + assert.Equal(t, 0, readSrc.callCount(), "preloaded key should be served without hitting the read source") +} + +func newLRUFetcher(t *testing.T, src Source[string], clk clock.Clock, ttl time.Duration, negatives *NegativeStore[string]) *Fetcher[string, string] { + t.Helper() + cache, err := NewLRUCache[string, string](100, clk) + require.NoError(t, err) + return New(Params[string, string]{ + Source: src, + Transform: identityTransform, + Cache: cache, + TTL: ttl, + Negatives: negatives, + Clock: clk, + }) +} + +func newServeStaleFetcher(t *testing.T, src Source[string], clk clock.Clock, ttl time.Duration) *Fetcher[string, string] { + t.Helper() + cache, err := NewLRUCache[string, string](100, clk) + require.NoError(t, err) + return New(Params[string, string]{ + Source: src, + Transform: identityTransform, + Cache: cache, + TTL: ttl, + ServeStale: true, + Clock: clk, + }) +} + +// TestGetExpiresAndReloadsByDefault verifies the default (serve-stale off): past +// TTL the entry is treated as expired and reloaded synchronously on the next read. +func TestGetExpiresAndReloadsByDefault(t *testing.T) { + clk := clock.NewMock() + src := &stubSource{data: map[string]json.RawMessage{"a": json.RawMessage(`v1`)}} + f := newLRUFetcher(t, src, clk, time.Hour, nil) + + _, err := f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, 1, src.callCount()) + + // Still fresh. + clk.Add(30 * time.Minute) + _, err = f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, 1, src.callCount()) + + // Past TTL with serve-stale off: the read reloads synchronously and returns fresh. + clk.Add(2 * time.Hour) + src.data["a"] = json.RawMessage(`v2`) + v, err := f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, "v2", v, "a stale entry must be reloaded synchronously, returning the fresh value") + assert.Equal(t, 2, src.callCount()) +} + +func TestGetHitAfterMiss(t *testing.T) { + src := &stubSource{data: map[string]json.RawMessage{"a": json.RawMessage(`v1`)}} + f := newLRUFetcher(t, src, clock.NewMock(), time.Hour, nil) + + v, err := f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, "v1", v) + + v, err = f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, "v1", v) + + assert.Equal(t, 1, src.callCount(), "second Get should be served from cache") +} + +func TestGetNotFoundWithNegativeCache(t *testing.T) { + clk := clock.NewMock() + neg, err := NewNegativeStore[string](10, time.Minute, clk) + require.NoError(t, err) + src := &stubSource{data: map[string]json.RawMessage{}} + f := newLRUFetcher(t, src, clk, time.Hour, neg) + + _, err = f.Get(context.Background(), "missing") + assert.ErrorIs(t, err, ErrNotFound) + + _, err = f.Get(context.Background(), "missing") + assert.ErrorIs(t, err, ErrNotFound) + + assert.Equal(t, 1, src.callCount(), "negative cache should prevent a second backend call") +} + +func TestGetNotFoundWithoutNegativeCache(t *testing.T) { + src := &stubSource{data: map[string]json.RawMessage{}} + f := newLRUFetcher(t, src, clock.NewMock(), time.Hour, nil) + + _, err := f.Get(context.Background(), "missing") + assert.ErrorIs(t, err, ErrNotFound) + _, err = f.Get(context.Background(), "missing") + assert.ErrorIs(t, err, ErrNotFound) + + assert.Equal(t, 2, src.callCount(), "without negative cache each miss hits the backend") +} + +func TestGetCoalescesConcurrentMisses(t *testing.T) { + src := &stubSource{ + data: map[string]json.RawMessage{"a": json.RawMessage(`v1`)}, + block: make(chan struct{}), + } + cache, err := NewLRUCache[string, string](100, clock.NewMock()) + require.NoError(t, err) + f := New(Params[string, string]{ + Source: src, + Transform: identityTransform, + Cache: cache, + TTL: time.Hour, + Coalesce: true, + }) + + const n = 8 + var wg sync.WaitGroup + results := make([]string, n) + for i := 0; i < n; i++ { + wg.Add(1) + go func(idx int) { + defer wg.Done() + v, err := f.Get(context.Background(), "a") + assert.NoError(t, err) + results[idx] = v + }(i) + } + + // Give the goroutines time to converge on the single in-flight call, then release it. + time.Sleep(50 * time.Millisecond) + close(src.block) + wg.Wait() + + for _, r := range results { + assert.Equal(t, "v1", r) + } + assert.Equal(t, 1, src.callCount(), "concurrent misses should collapse into a single backend call") +} + +func TestGetServesStaleAndRefreshesInBackground(t *testing.T) { + clk := clock.NewMock() + src := &stubSource{data: map[string]json.RawMessage{"a": json.RawMessage(`v1`)}} + f := newServeStaleFetcher(t, src, clk, time.Hour) + + // Cold load. + v, err := f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, "v1", v) + assert.Equal(t, 1, src.callCount()) + + // Still fresh: no extra fetch. + clk.Add(30 * time.Minute) + _, err = f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, 1, src.callCount()) + + // Past TTL: the read returns the stale value immediately and refreshes in the + // background (never blocks). The backend value changes so we can observe it. + clk.Add(2 * time.Hour) + src.data["a"] = json.RawMessage(`v2`) + v, err = f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, "v1", v, "stale read must return the last good value, not block") + + // The background refresh eventually re-fetches and updates the cache. + assert.Eventually(t, func() bool { return src.callCount() == 2 }, time.Second, 5*time.Millisecond, + "a stale read should trigger exactly one background refresh") + assert.Eventually(t, func() bool { + got, _ := f.Get(context.Background(), "a") + return got == "v2" + }, time.Second, 5*time.Millisecond, "the refreshed value should become visible") +} + +func TestGetServesStaleWhileBackendDown(t *testing.T) { + clk := clock.NewMock() + src := &stubSource{data: map[string]json.RawMessage{"a": json.RawMessage(`v1`)}} + f := newServeStaleFetcher(t, src, clk, time.Hour) + + // Warm the cache. + _, err := f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, 1, src.callCount()) + + // Backend goes down and the entry goes stale. + src.err = errors.New("backend down") + clk.Add(2 * time.Hour) + + // Reads keep returning the last good value; failed refresh backs off so the + // backend is not hammered. + for i := 0; i < 5; i++ { + v, gErr := f.Get(context.Background(), "a") + require.NoError(t, gErr) + assert.Equal(t, "v1", v) + } + assert.Eventually(t, func() bool { return src.callCount() >= 2 }, time.Second, 5*time.Millisecond) + assert.LessOrEqual(t, src.callCount(), 2, "failed refreshes must back off, not storm the backend") +} + +func TestBackgroundRevalidationTimeoutReleasesSlot(t *testing.T) { + clk := clock.NewMock() + src := &timeoutOnceSource{data: map[string]json.RawMessage{"a": json.RawMessage(`v1`)}} + cache, err := NewLRUCache[string, string](100, clk) + require.NoError(t, err) + f := New(Params[string, string]{ + Source: src, + Transform: identityTransform, + Cache: cache, + TTL: time.Hour, + ServeStale: true, + RevalidateTimeout: 10 * time.Millisecond, + Clock: clk, + }) + + v, err := f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, "v1", v) + + clk.Add(2 * time.Hour) + src.data["a"] = json.RawMessage(`v2`) + v, err = f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, "v1", v) + assert.Eventually(t, func() bool { return src.callCount() == 2 }, time.Second, 5*time.Millisecond) + assert.Eventually(t, func() bool { + f.reval.mu.Lock() + defer f.reval.mu.Unlock() + st := f.reval.state["a"] + return !st.inFlight && !st.failedAt.IsZero() + }, time.Second, 5*time.Millisecond) + + clk.Add(revalidateBackoff + time.Second) + v, err = f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, "v1", v) + assert.Eventually(t, func() bool { return src.callCount() == 3 }, time.Second, 5*time.Millisecond) + assert.Eventually(t, func() bool { + got, _ := f.Get(context.Background(), "a") + return got == "v2" + }, time.Second, 5*time.Millisecond) +} + +func TestRevalidatorPrunesExpiredFailures(t *testing.T) { + clk := clock.NewMock() + r := newRevalidator[string](clk, revalidateBackoff) + r.finish("old", true) + require.Contains(t, r.state, "old") + + clk.Add(revalidateBackoff + time.Second) + assert.True(t, r.begin("new")) + assert.NotContains(t, r.state, "old") + assert.True(t, r.state["new"].inFlight) +} + +func TestGetNoCacheAlwaysFetches(t *testing.T) { + src := &stubSource{data: map[string]json.RawMessage{"a": json.RawMessage(`v1`)}} + f := New(Params[string, string]{ + Source: src, + Transform: identityTransform, + Cache: NoCache[string, string]{}, + TTL: time.Hour, + }) + + for i := 0; i < 3; i++ { + v, err := f.Get(context.Background(), "a") + require.NoError(t, err) + assert.Equal(t, "v1", v) + } + assert.Equal(t, 3, src.callCount()) +} + +func TestGetSourceErrorNotCached(t *testing.T) { + boom := errors.New("boom") + src := &stubSource{err: boom} + f := newLRUFetcher(t, src, clock.NewMock(), time.Hour, nil) + + _, err := f.Get(context.Background(), "a") + assert.ErrorIs(t, err, boom) + assert.NotErrorIs(t, err, ErrNotFound) + + _, err = f.Get(context.Background(), "a") + assert.ErrorIs(t, err, boom) + assert.Equal(t, 2, src.callCount(), "systemic errors must not be cached") +} + +func TestGetTransformErrorNotCached(t *testing.T) { + src := &stubSource{data: map[string]json.RawMessage{"a": json.RawMessage(`v1`)}} + cache, err := NewLRUCache[string, string](100, clock.NewMock()) + require.NoError(t, err) + transformErr := errors.New("bad value") + f := New(Params[string, string]{ + Source: src, + Transform: func(string, json.RawMessage) (string, error) { return "", transformErr }, + Cache: cache, + TTL: time.Hour, + }) + + _, err = f.Get(context.Background(), "a") + assert.ErrorIs(t, err, transformErr) + _, err = f.Get(context.Background(), "a") + assert.ErrorIs(t, err, transformErr) + assert.Equal(t, 2, src.callCount(), "malformed values must not be cached") +} + +// countingRecorder verifies the engine emits the expected telemetry. +type countingRecorder struct { + hits, misses, negatives int + backend map[string]int +} + +func newCountingRecorder() *countingRecorder { + return &countingRecorder{backend: map[string]int{}} +} + +func (r *countingRecorder) CacheHit() { r.hits++ } +func (r *countingRecorder) CacheMiss() { r.misses++ } +func (r *countingRecorder) CacheNegative() { r.negatives++ } +func (r *countingRecorder) BackendFetch(result string, _ time.Duration) { r.backend[result]++ } + +func TestRecorderSignals(t *testing.T) { + clk := clock.NewMock() + neg, err := NewNegativeStore[string](10, time.Minute, clk) + require.NoError(t, err) + cache, err := NewLRUCache[string, string](100, clk) + require.NoError(t, err) + rec := newCountingRecorder() + src := &stubSource{data: map[string]json.RawMessage{"a": json.RawMessage(`v1`)}} + f := New(Params[string, string]{ + Source: src, + Transform: identityTransform, + Cache: cache, + TTL: time.Hour, + Negatives: neg, + Clock: clk, + Metrics: rec, + }) + + // miss -> ok, then hit. + _, _ = f.Get(context.Background(), "a") + _, _ = f.Get(context.Background(), "a") + // miss -> notfound, then negative. + _, _ = f.Get(context.Background(), "missing") + _, _ = f.Get(context.Background(), "missing") + + assert.Equal(t, 1, rec.hits) + assert.Equal(t, 2, rec.misses) + assert.Equal(t, 1, rec.negatives) + assert.Equal(t, 1, rec.backend["ok"]) + assert.Equal(t, 1, rec.backend["notfound"]) +} diff --git a/cachekit/contracts.go b/cachekit/contracts.go new file mode 100644 index 00000000000..61cbede5cf9 --- /dev/null +++ b/cachekit/contracts.go @@ -0,0 +1,63 @@ +package cachekit + +import ( + "context" + "encoding/json" + "errors" + "time" +) + +// ErrNotFound is returned by Get when the key does not exist at the source. +// It is a definitive, per-key verdict (as opposed to a systemic/backend error). +var ErrNotFound = errors.New("cachekit: not found") + +// Source pulls raw, undecoded bytes for a batch of keys. A single-key lookup is +// a one-element slice. By convention, a key that is absent from the returned map +// is treated as a definitive "not found" for that key; a non-nil error is a +// systemic failure (never cached, never negative-cached). +type Source[K comparable] interface { + Fetch(ctx context.Context, keys []K) (map[K]json.RawMessage, error) +} + +// BulkSource is an optional capability a Source may implement to return the entire +// corpus in a single call. It is what the engine's preload warm-up uses to fill the +// cache at startup; sources that cannot enumerate everything simply do not implement it. +type BulkSource[K comparable] interface { + FetchAll(ctx context.Context) (map[K]json.RawMessage, error) +} + +// TransformFunc converts raw bytes into the typed value V. It runs exactly once +// per key, at cache insert time. The key is provided so transforms that need it +// (e.g. filling an ID) can do so without mutating the shared cached value later. +type TransformFunc[K comparable, V any] func(key K, raw json.RawMessage) (V, error) + +// Cache is a keyed store of composed typed values. Implementations must be safe +// for concurrent use. +type Cache[K comparable, V any] interface { + // Get returns the value if present, and whether it is stale (past its refresh + // time). Stale values are still returned so the read path never blocks on the + // backend; the engine revalidates them in the background. + Get(key K) (v V, ok bool, stale bool) + Save(key K, v V, ttl time.Duration) + // Invalidate drops a key so the next Get is a miss. Used when a background + // revalidation finds the key was deleted upstream. + Invalidate(key K) +} + +// Recorder receives low-cardinality telemetry. The subsystem label is applied by +// the implementation, not passed per call, to keep cardinality bounded. +type Recorder interface { + CacheHit() + CacheMiss() + CacheNegative() + // BackendFetch reports one upstream call. result is "ok", "notfound" or "error". + BackendFetch(result string, d time.Duration) +} + +// noopRecorder is the default Recorder. +type noopRecorder struct{} + +func (noopRecorder) CacheHit() {} +func (noopRecorder) CacheMiss() {} +func (noopRecorder) CacheNegative() {} +func (noopRecorder) BackendFetch(string, time.Duration) {} diff --git a/cachekit/negative.go b/cachekit/negative.go new file mode 100644 index 00000000000..077edf923f8 --- /dev/null +++ b/cachekit/negative.go @@ -0,0 +1,59 @@ +package cachekit + +import ( + "time" + + "github.com/benbjohnson/clock" + lru "github.com/hashicorp/golang-lru/v2" +) + +// negativeEntry is a cached definitive verdict: the error to re-serve and its +// expiry. The error is preserved (rather than collapsed to a generic marker) so +// callers see the real verdict — e.g. ErrNotFound vs a malformed-value error. +type negativeEntry struct { + err error + expires time.Time +} + +// NegativeStore is a small, bounded store of definitive verdicts (not-found or +// permanent malformed). It is kept separate from the positive cache with its own +// capacity and short TTL so an unknown-key flood can never evict real data. Only +// permanent verdicts belong here; transient failures are never stored. Safe for +// concurrent use. +type NegativeStore[K comparable] struct { + lru *lru.Cache[K, negativeEntry] + ttl time.Duration + clock clock.Clock +} + +// NewNegativeStore builds a negative store holding up to maxEntries verdicts, each +// retained for ttl. +func NewNegativeStore[K comparable](maxEntries int, ttl time.Duration, clk clock.Clock) (*NegativeStore[K], error) { + if clk == nil { + clk = clock.New() + } + l, err := lru.New[K, negativeEntry](maxEntries) + if err != nil { + return nil, err + } + return &NegativeStore[K]{lru: l, ttl: ttl, clock: clk}, nil +} + +// isCached reports whether key has a live negative verdict, returning the verdict +// error to re-serve (e.g. ErrNotFound or a malformed-value error). +func (n *NegativeStore[K]) isCached(key K) (error, bool) { + e, ok := n.lru.Get(key) + if !ok { + return nil, false + } + if n.clock.Now().After(e.expires) { + n.lru.Remove(key) + return nil, false + } + return e.err, true +} + +// mark records a definitive verdict error for key, retained for the store TTL. +func (n *NegativeStore[K]) mark(key K, err error) { + n.lru.Add(key, negativeEntry{err: err, expires: n.clock.Now().Add(n.ttl)}) +} diff --git a/cachekit/revalidate.go b/cachekit/revalidate.go new file mode 100644 index 00000000000..ef9cabf13d6 --- /dev/null +++ b/cachekit/revalidate.go @@ -0,0 +1,120 @@ +package cachekit + +import ( + "context" + "sync" + "time" + + "github.com/benbjohnson/clock" +) + +// revalidateBackoff is how long a key waits after a failed background revalidation +// before another is attempted, so a struggling backend is not hammered. +const revalidateBackoff = 5 * time.Second + +// revalState is a key's background-revalidation state: whether one is in flight +// and, if the last one failed, when (so the next attempt can back off). +type revalState struct { + inFlight bool + failedAt time.Time +} + +// revalidator serialises background revalidations per key — at most one in flight, +// with a backoff after a failure. It owns its own lock so callers never touch it. +type revalidator[K comparable] struct { + mu sync.Mutex + state map[K]revalState + backoff time.Duration + clock clock.Clock +} + +func newRevalidator[K comparable](clk clock.Clock, backoff time.Duration) *revalidator[K] { + return &revalidator[K]{state: make(map[K]revalState), backoff: backoff, clock: clk} +} + +// begin reports whether the caller may start a revalidation for key now, and claims +// the in-flight slot if so. It returns false when one is already running or the last +// attempt failed within the backoff window. +func (r *revalidator[K]) begin(key K) bool { + r.mu.Lock() + defer r.mu.Unlock() + now := r.clock.Now() + r.pruneExpiredFailures(now) + st := r.state[key] + if st.inFlight || (!st.failedAt.IsZero() && r.clock.Now().Before(st.failedAt.Add(r.backoff))) { + return false + } + st.inFlight = true + r.state[key] = st + return true +} + +func (r *revalidator[K]) pruneExpiredFailures(now time.Time) { + for key, st := range r.state { + if !st.inFlight && !st.failedAt.IsZero() && !now.Before(st.failedAt.Add(r.backoff)) { + delete(r.state, key) + } + } +} + +// finish releases the in-flight slot; on failure it records the time (for backoff), +// on success it forgets the key entirely. +func (r *revalidator[K]) finish(key K, failed bool) { + r.mu.Lock() + defer r.mu.Unlock() + if failed { + r.state[key] = revalState{failedAt: r.clock.Now()} + } else { + delete(r.state, key) + } +} + +// triggerRevalidate starts one background revalidation for a stale key. It never +// blocks the caller: the revalidator admits at most one per key at a time and backs +// off after failures, so a struggling backend is not hammered and stale values keep +// being served. +func (f *Fetcher[K, V]) triggerRevalidate(key K) { + if !f.reval.begin(key) { + return + } + ctx, cancel := context.WithTimeout(context.Background(), f.revalTimeout) + go f.revalidate(ctx, cancel, key) +} + +// revalidate reloads a stale key in the background. It never worsens availability: +// on success it replaces the value; if the key is gone upstream it is dropped (and +// negative-cached); on any error (transient or a newly-malformed value) the last +// good value keeps being served and a backoff is recorded. +func (f *Fetcher[K, V]) revalidate(ctx context.Context, cancel context.CancelFunc, key K) { + defer cancel() + start := f.clock.Now() + found, err := f.source.Fetch(ctx, []K{key}) + dur := f.clock.Now().Sub(start) + + if err != nil { + f.metrics.BackendFetch("error", dur) + f.reval.finish(key, true) + return + } + raw, ok := found[key] + if !ok { + // Deleted upstream: drop it so the next read reflects the deletion. + f.metrics.BackendFetch("notfound", dur) + f.cache.Invalidate(key) + if f.negatives != nil { + f.negatives.mark(key, ErrNotFound) + } + f.reval.finish(key, false) + return + } + v, err := f.transform(key, raw) + if err != nil { + // Newly-malformed upstream value: keep serving the last good value. + f.metrics.BackendFetch("error", dur) + f.reval.finish(key, true) + return + } + f.cache.Save(key, v, f.ttl) + f.metrics.BackendFetch("ok", dur) + f.reval.finish(key, false) +} diff --git a/config/config.go b/config/config.go index 5d4bc863776..5bdea632885 100644 --- a/config/config.go +++ b/config/config.go @@ -1148,6 +1148,18 @@ func SetupViper(v *viper.Viper, filename string, bidderInfos BidderInfos) { v.SetDefault("accounts.http_events.endpoint", "") v.SetDefault("accounts.http_events.refresh_rate_seconds", 0) v.SetDefault("accounts.http_events.timeout_ms", 0) + // Fetchers 2.0 (cachekit) account cache. Opt-in via accounts.v2_enabled; the + // defaults below only take effect when it is turned on. + v.SetDefault("accounts.v2_enabled", false) + v.SetDefault("accounts.cache.type", "lru") + v.SetDefault("accounts.cache.max_entries", 50000) + v.SetDefault("accounts.cache.ttl_seconds", 3600) + v.SetDefault("accounts.cache.refresh", "ttl") + v.SetDefault("accounts.cache.coalesce_requests", false) + v.SetDefault("accounts.cache.serve_stale", false) + v.SetDefault("accounts.cache.negative.enabled", false) + v.SetDefault("accounts.cache.negative.max_entries", 10000) + v.SetDefault("accounts.cache.negative.ttl_seconds", 60) v.BindEnv("user_sync.external_url") v.BindEnv("user_sync.coop_sync.default") diff --git a/config/stored_requests.go b/config/stored_requests.go index 92eab4ea8e8..26f9f9676bd 100644 --- a/config/stored_requests.go +++ b/config/stored_requests.go @@ -71,6 +71,79 @@ type StoredRequests struct { // HTTPEvents configures an instance of stored_requests/events/http/http.go. // If non-nil, the server will use those endpoints to populate and update the cache. HTTPEvents HTTPEventsConfig `mapstructure:"http_events"` + // V2Enabled opts this data type into the Fetchers 2.0 (cachekit) read path. + // Currently honoured for the accounts data type only. Defaults to false, which + // preserves the legacy fetch + byte-cache behaviour unchanged. + V2Enabled bool `mapstructure:"v2_enabled"` + // CacheV2 configures the Fetchers 2.0 typed cache. Only used when V2Enabled is true. + CacheV2 CacheKitConfig `mapstructure:"cache"` +} + +// CacheKitConfig configures a Fetchers 2.0 typed cache (cachekit). +type CacheKitConfig struct { + // Type selects the cache retention policy: "none" (always fetch from source) or + // "lru" (bounded read-through, the default). + Type string `mapstructure:"type"` + // MaxEntries bounds the number of cached values when Type is "lru". + MaxEntries int `mapstructure:"max_entries"` + // TTLSeconds is how long a cached value is served fresh before a background + // refresh is triggered. Past it the value is still served (stale) while it + // refreshes, so reads never block on the backend. + TTLSeconds int `mapstructure:"ttl_seconds"` + // Refresh selects the freshness mode: "ttl" (serve-stale + background refresh, + // the default), "none" (never refresh / load-once) or "preload" (bulk warm at + // startup then ttl). + Refresh RefreshMode `mapstructure:"refresh"` + // CoalesceRequests opts into single-flight coalescing so concurrent misses for + // the same key collapse into one upstream fetch. Opt-in; defaults to off. + CoalesceRequests bool `mapstructure:"coalesce_requests"` + // ServeStale opts into stale-while-revalidate: past ttl_seconds the cached value + // is served immediately and refreshed in the background (reads never block on the + // backend). For refresh modes "ttl" and "preload", this behavior is enabled by + // default to match the mode contract. + ServeStale bool `mapstructure:"serve_stale"` + // RevalidateTimeoutSeconds is the maximum time allowed for background stale + // revalidation before the attempt is failed and retried after backoff. + RevalidateTimeoutSeconds int `mapstructure:"revalidate_timeout_seconds"` + // Negative configures the optional negative (definitive-verdict) cache. + Negative NegativeCacheConfig `mapstructure:"negative"` +} + +// RefreshMode selects a cachekit freshness mode. +type RefreshMode string + +const ( + // RefreshTTL serves stale values while refreshing them in the background. + RefreshTTL RefreshMode = "ttl" + // RefreshNone never refreshes cached values (load-once / mirror). + RefreshNone RefreshMode = "none" + // RefreshPreload bulk-warms the cache at startup, then behaves like ttl. + RefreshPreload RefreshMode = "preload" +) + +// TTL returns the positive-cache time-to-live. +func (c CacheKitConfig) TTL() time.Duration { + return time.Duration(c.TTLSeconds) * time.Second +} + +// RevalidateTimeout returns the background revalidation timeout. +func (c CacheKitConfig) RevalidateTimeout() time.Duration { + return time.Duration(c.RevalidateTimeoutSeconds) * time.Second +} + +// NegativeCacheConfig configures caching of definitive not-found verdicts. +type NegativeCacheConfig struct { + // Enabled turns negative caching on. Opt-in; defaults to off. + Enabled bool `mapstructure:"enabled"` + // MaxEntries bounds the negative store, kept separate from the positive cache. + MaxEntries int `mapstructure:"max_entries"` + // TTLSeconds is how long a not-found verdict is retained. + TTLSeconds int `mapstructure:"ttl_seconds"` +} + +// TTL returns the negative-cache time-to-live. +func (c NegativeCacheConfig) TTL() time.Duration { + return time.Duration(c.TTLSeconds) * time.Second } // HTTPEventsConfig configures stored_requests/events/http/http.go diff --git a/go.mod b/go.mod index 363683694e4..73e20fadc25 100644 --- a/go.mod +++ b/go.mod @@ -19,6 +19,7 @@ require ( github.com/gofrs/uuid v4.2.0+incompatible github.com/golang/glog v1.2.5 github.com/google/go-cmp v0.7.0 + github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/json-iterator/go v1.1.12 github.com/julienschmidt/httprouter v1.3.0 github.com/lib/pq v1.10.4 @@ -41,6 +42,7 @@ require ( github.com/xeipuuv/gojsonschema v1.2.0 github.com/yudai/gojsondiff v1.0.0 golang.org/x/net v0.55.0 + golang.org/x/sync v0.20.0 golang.org/x/text v0.37.0 google.golang.org/grpc v1.79.3 gopkg.in/evanphx/json-patch.v5 v5.9.0 diff --git a/go.sum b/go.sum index 6184dea4d16..6d9ee6edc95 100644 --- a/go.sum +++ b/go.sum @@ -285,6 +285,8 @@ github.com/hashicorp/go-version v1.1.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64= @@ -684,6 +686,8 @@ golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= diff --git a/metrics/config/metrics.go b/metrics/config/metrics.go index 6ef2d552c0c..d2d379430e2 100644 --- a/metrics/config/metrics.go +++ b/metrics/config/metrics.go @@ -238,6 +238,20 @@ func (me *MultiMetricsEngine) RecordAccountCacheResult(cacheResult metrics.Cache } } +// RecordCacheKitResult across all engines +func (me *MultiMetricsEngine) RecordCacheKitResult(subsystem string, result metrics.CacheKitResult) { + for _, thisME := range *me { + thisME.RecordCacheKitResult(subsystem, result) + } +} + +// RecordCacheKitBackendFetch across all engines +func (me *MultiMetricsEngine) RecordCacheKitBackendFetch(subsystem string, result metrics.CacheKitBackendResult, length time.Duration) { + for _, thisME := range *me { + thisME.RecordCacheKitBackendFetch(subsystem, result, length) + } +} + // RecordPrebidCacheRequestTime across all engines func (me *MultiMetricsEngine) RecordPrebidCacheRequestTime(success bool, length time.Duration) { for _, thisME := range *me { @@ -506,6 +520,14 @@ func (me *NilMetricsEngine) RecordStoredImpCacheResult(cacheResult metrics.Cache func (me *NilMetricsEngine) RecordAccountCacheResult(cacheResult metrics.CacheResult, inc int) { } +// RecordCacheKitResult as a noop +func (me *NilMetricsEngine) RecordCacheKitResult(subsystem string, result metrics.CacheKitResult) { +} + +// RecordCacheKitBackendFetch as a noop +func (me *NilMetricsEngine) RecordCacheKitBackendFetch(subsystem string, result metrics.CacheKitBackendResult, length time.Duration) { +} + // RecordPrebidCacheRequestTime as a noop func (me *NilMetricsEngine) RecordPrebidCacheRequestTime(success bool, length time.Duration) { } diff --git a/metrics/go_metrics.go b/metrics/go_metrics.go index 4f10166c171..26a738a850d 100644 --- a/metrics/go_metrics.go +++ b/metrics/go_metrics.go @@ -937,6 +937,14 @@ func (me *Metrics) RecordAccountCacheResult(cacheResult CacheResult, inc int) { me.AccountCacheMeter[cacheResult].Mark(int64(inc)) } +// RecordCacheKitResult is not tracked by the influx/go-metrics engine. +func (me *Metrics) RecordCacheKitResult(subsystem string, result CacheKitResult) { +} + +// RecordCacheKitBackendFetch is not tracked by the influx/go-metrics engine. +func (me *Metrics) RecordCacheKitBackendFetch(subsystem string, result CacheKitBackendResult, length time.Duration) { +} + // RecordPrebidCacheRequestTime implements a part of the MetricsEngine interface. Records the // amount of time taken to store the auction result in Prebid Cache. func (me *Metrics) RecordPrebidCacheRequestTime(success bool, length time.Duration) { diff --git a/metrics/metrics.go b/metrics/metrics.go index aa2e28b7c8c..2cec42abe5c 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -340,6 +340,48 @@ func CacheResults() []CacheResult { } } +// CacheKitResult is the outcome of a Fetchers 2.0 (cachekit) cache lookup. +type CacheKitResult string + +const ( + // CacheKitResultHit is a value served from the positive cache without a backend call. + CacheKitResultHit CacheKitResult = "hit" + // CacheKitResultMiss is a lookup that had to go to the backend source. + CacheKitResultMiss CacheKitResult = "miss" + // CacheKitResultNegative is a not-found served from the negative cache without a backend call. + CacheKitResultNegative CacheKitResult = "negative" +) + +// CacheKitResults returns the possible cachekit cache lookup outcomes. +func CacheKitResults() []CacheKitResult { + return []CacheKitResult{ + CacheKitResultHit, + CacheKitResultMiss, + CacheKitResultNegative, + } +} + +// CacheKitBackendResult is the outcome of a cachekit upstream (source) fetch. +type CacheKitBackendResult string + +const ( + // CacheKitBackendOK is a successful backend fetch that returned a value. + CacheKitBackendOK CacheKitBackendResult = "ok" + // CacheKitBackendNotFound is a definitive per-key not-found from the backend. + CacheKitBackendNotFound CacheKitBackendResult = "notfound" + // CacheKitBackendError is a systemic backend failure (never cached). + CacheKitBackendError CacheKitBackendResult = "error" +) + +// CacheKitBackendResults returns the possible cachekit backend fetch outcomes. +func CacheKitBackendResults() []CacheKitBackendResult { + return []CacheKitBackendResult{ + CacheKitBackendOK, + CacheKitBackendNotFound, + CacheKitBackendError, + } +} + // TCFVersionValue : The possible values for TCF versions type TCFVersionValue string @@ -488,6 +530,8 @@ type MetricsEngine interface { RecordAccountCacheResult(cacheResult CacheResult, inc int) RecordStoredDataFetchTime(labels StoredDataLabels, length time.Duration) RecordStoredDataError(labels StoredDataLabels) + RecordCacheKitResult(subsystem string, result CacheKitResult) + RecordCacheKitBackendFetch(subsystem string, result CacheKitBackendResult, length time.Duration) RecordPrebidCacheRequestTime(success bool, length time.Duration) RecordRequestQueueTime(success bool, requestType RequestType, length time.Duration) RecordTimeoutNotice(success bool) diff --git a/metrics/metrics_mock.go b/metrics/metrics_mock.go index 7cf0af5cc0b..b0b91854d0e 100644 --- a/metrics/metrics_mock.go +++ b/metrics/metrics_mock.go @@ -136,6 +136,14 @@ func (me *MetricsEngineMock) RecordAccountCacheResult(cacheResult CacheResult, i me.Called(cacheResult, inc) } +func (me *MetricsEngineMock) RecordCacheKitResult(subsystem string, result CacheKitResult) { + me.Called(subsystem, result) +} + +func (me *MetricsEngineMock) RecordCacheKitBackendFetch(subsystem string, result CacheKitBackendResult, length time.Duration) { + me.Called(subsystem, result, length) +} + // RecordPrebidCacheRequestTime mock func (me *MetricsEngineMock) RecordPrebidCacheRequestTime(success bool, length time.Duration) { me.Called(success, length) diff --git a/metrics/prometheus/prometheus.go b/metrics/prometheus/prometheus.go index cd694b67c85..04c921e2883 100644 --- a/metrics/prometheus/prometheus.go +++ b/metrics/prometheus/prometheus.go @@ -62,6 +62,11 @@ type Metrics struct { adsCertSignTimer prometheus.Histogram bidderServerResponseTimer prometheus.Histogram + // Fetchers 2.0 (cachekit) Metrics + cacheKitResult *prometheus.CounterVec + cacheKitBackendFetch *prometheus.CounterVec + cacheKitBackendFetchTimer *prometheus.HistogramVec + // Adapter Metrics adapterBids *prometheus.CounterVec adapterErrors *prometheus.CounterVec @@ -116,6 +121,7 @@ const ( adapterLabel = "adapter" bidTypeLabel = "bid_type" cacheResultLabel = "cache_result" + cacheKitResultLabel = "result" connectionErrorLabel = "connection_error" cookieLabel = "cookie" hasBidsLabel = "has_bids" @@ -133,6 +139,7 @@ const ( stageLabel = "stage" statusLabel = "status" successLabel = "success" + subsystemLabel = "subsystem" syncerLabel = "syncer" versionLabel = "version" ) @@ -269,6 +276,22 @@ func NewMetrics(cfg config.PrometheusMetrics, disabledMetrics config.DisabledMet "Count of stored account errors by error type", []string{storedDataErrorLabel}) + metrics.cacheKitResult = newCounter(cfg, reg, + "cachekit_cache_result", + "Count of Fetchers 2.0 cache lookups labeled by subsystem and result (hit, miss, negative).", + []string{subsystemLabel, cacheKitResultLabel}) + + metrics.cacheKitBackendFetch = newCounter(cfg, reg, + "cachekit_backend_fetch", + "Count of Fetchers 2.0 backend fetches labeled by subsystem and result (ok, notfound, error).", + []string{subsystemLabel, cacheKitResultLabel}) + + metrics.cacheKitBackendFetchTimer = newHistogramVec(cfg, reg, + "cachekit_backend_fetch_duration_seconds", + "Seconds to fetch a value from a Fetchers 2.0 backend source, labeled by subsystem.", + []string{subsystemLabel}, + standardTimeBuckets) + metrics.storedAMPFetchTimer = newHistogramVec(cfg, reg, "stored_amp_fetch_time_seconds", "Seconds to fetch stored AMP requests labeled by fetch type", @@ -965,6 +988,23 @@ func (m *Metrics) RecordAccountCacheResult(cacheResult metrics.CacheResult, inc }).Add(float64(inc)) } +func (m *Metrics) RecordCacheKitResult(subsystem string, result metrics.CacheKitResult) { + m.cacheKitResult.With(prometheus.Labels{ + subsystemLabel: subsystem, + cacheKitResultLabel: string(result), + }).Inc() +} + +func (m *Metrics) RecordCacheKitBackendFetch(subsystem string, result metrics.CacheKitBackendResult, length time.Duration) { + m.cacheKitBackendFetch.With(prometheus.Labels{ + subsystemLabel: subsystem, + cacheKitResultLabel: string(result), + }).Inc() + m.cacheKitBackendFetchTimer.With(prometheus.Labels{ + subsystemLabel: subsystem, + }).Observe(length.Seconds()) +} + func (m *Metrics) RecordPrebidCacheRequestTime(success bool, length time.Duration) { m.prebidCacheWriteTimer.With(prometheus.Labels{ successLabel: strconv.FormatBool(success), diff --git a/metrics/prometheus/prometheus_test.go b/metrics/prometheus/prometheus_test.go index c9ca024ef96..61ea5a3df13 100644 --- a/metrics/prometheus/prometheus_test.go +++ b/metrics/prometheus/prometheus_test.go @@ -1224,6 +1224,44 @@ func TestAccountCacheResultMetric(t *testing.T) { }) } +func TestCacheKitMetrics(t *testing.T) { + m := createMetricsForTesting() + + m.RecordCacheKitResult("account", metrics.CacheKitResultHit) + m.RecordCacheKitResult("account", metrics.CacheKitResultHit) + m.RecordCacheKitResult("account", metrics.CacheKitResultMiss) + m.RecordCacheKitResult("account", metrics.CacheKitResultNegative) + + m.RecordCacheKitBackendFetch("account", metrics.CacheKitBackendOK, time.Millisecond) + m.RecordCacheKitBackendFetch("account", metrics.CacheKitBackendNotFound, time.Millisecond) + + assertCounterVecValue(t, "", "cacheKitResult:hit", m.cacheKitResult, 2, + prometheus.Labels{ + subsystemLabel: "account", + cacheKitResultLabel: string(metrics.CacheKitResultHit), + }) + assertCounterVecValue(t, "", "cacheKitResult:miss", m.cacheKitResult, 1, + prometheus.Labels{ + subsystemLabel: "account", + cacheKitResultLabel: string(metrics.CacheKitResultMiss), + }) + assertCounterVecValue(t, "", "cacheKitResult:negative", m.cacheKitResult, 1, + prometheus.Labels{ + subsystemLabel: "account", + cacheKitResultLabel: string(metrics.CacheKitResultNegative), + }) + assertCounterVecValue(t, "", "cacheKitBackendFetch:ok", m.cacheKitBackendFetch, 1, + prometheus.Labels{ + subsystemLabel: "account", + cacheKitResultLabel: string(metrics.CacheKitBackendOK), + }) + assertCounterVecValue(t, "", "cacheKitBackendFetch:notfound", m.cacheKitBackendFetch, 1, + prometheus.Labels{ + subsystemLabel: "account", + cacheKitResultLabel: string(metrics.CacheKitBackendNotFound), + }) +} + func TestCookieSyncMetric(t *testing.T) { tests := []struct { status metrics.CookieSyncStatus diff --git a/stored_requests/backends/db_fetcher/fetcher.go b/stored_requests/backends/db_fetcher/fetcher.go index 13c5cb00b98..1d9ff2981ca 100644 --- a/stored_requests/backends/db_fetcher/fetcher.go +++ b/stored_requests/backends/db_fetcher/fetcher.go @@ -1,8 +1,10 @@ package db_fetcher import ( + "bytes" "context" "encoding/json" + "errors" "github.com/lib/pq" "github.com/prebid/prebid-server/v4/logger" @@ -15,6 +17,18 @@ func NewFetcher( queryTemplate string, responseQueryTemplate string, ) stored_requests.AllFetcher { + return NewFetcherWithAccountsQuery(provider, queryTemplate, responseQueryTemplate, "") +} + +// NewFetcherWithAccountsQuery is like NewFetcher but also accepts a query that +// returns every account row (id, data, dataType), enabling FetchAllAccounts for +// bulk cache preloading. An empty accountsQuery disables bulk account loading. +func NewFetcherWithAccountsQuery( + provider db_provider.DbProvider, + queryTemplate string, + responseQueryTemplate string, + accountsQuery string, +) stored_requests.AllFetcher { if provider == nil { logger.Fatalf("The Database Stored Request Fetcher requires a database connection. Please report this as a bug.") @@ -29,6 +43,7 @@ func NewFetcher( provider: provider, queryTemplate: queryTemplate, responseQueryTemplate: responseQueryTemplate, + accountsQuery: accountsQuery, } } @@ -37,6 +52,7 @@ type dbFetcher struct { provider db_provider.DbProvider queryTemplate string responseQueryTemplate string + accountsQuery string } func (fetcher *dbFetcher) FetchRequests(ctx context.Context, requestIDs []string, impIDs []string) (map[string]json.RawMessage, map[string]json.RawMessage, []error) { @@ -154,6 +170,44 @@ func (fetcher *dbFetcher) FetchAccount(ctx context.Context, accountDefaultsJSON return nil, []error{stored_requests.NotFoundError{ID: accountID, DataType: "Account"}} } +// FetchAllAccounts runs the configured accounts query and returns every account +// row keyed by ID. Rows with null/empty data are skipped. The bytes are raw (not +// defaults-merged); callers merge account defaults as needed. Bulk loading is +// unavailable when no accounts query was configured. +func (fetcher *dbFetcher) FetchAllAccounts(ctx context.Context) (map[string]json.RawMessage, []error) { + if fetcher.accountsQuery == "" { + return nil, []error{errors.New("db_fetcher: bulk account loading is not configured (set stored_requests.database.initialize_caches.query)")} + } + + rows, err := fetcher.provider.QueryContext(ctx, fetcher.accountsQuery) + if err != nil { + return nil, []error{err} + } + defer func() { + if cerr := rows.Close(); cerr != nil { + logger.Errorf("error closing DB connection: %v", cerr) + } + }() + + accounts := make(map[string]json.RawMessage) + for rows.Next() { + var id string + var data []byte + var dataType string + if err := rows.Scan(&id, &data, &dataType); err != nil { + return nil, []error{err} + } + if len(data) == 0 || bytes.Equal(data, []byte("null")) { + continue + } + accounts[id] = data + } + if rows.Err() != nil { + return nil, []error{rows.Err()} + } + return accounts, nil +} + func (fetcher *dbFetcher) FetchCategories(ctx context.Context, primaryAdServer, publisherId, iabCategory string) (string, error) { return "", nil } diff --git a/stored_requests/backends/db_fetcher/fetcher_test.go b/stored_requests/backends/db_fetcher/fetcher_test.go index 1f4464bcb8c..544d83a5d37 100644 --- a/stored_requests/backends/db_fetcher/fetcher_test.go +++ b/stored_requests/backends/db_fetcher/fetcher_test.go @@ -37,6 +37,43 @@ func TestEmptyQuery(t *testing.T) { assertMapLength(t, 0, storedResponses) } +func TestFetchAllAccountsNotConfigured(t *testing.T) { + provider, _, err := db_provider.NewDbProviderMock() + if err != nil { + t.Fatalf("Unexpected error stubbing DB: %v", err) + } + defer provider.Close() + + fetcher := &dbFetcher{provider: provider} + accounts, errs := fetcher.FetchAllAccounts(context.Background()) + assert.Nil(t, accounts) + assertErrorCount(t, 1, errs) +} + +func TestFetchAllAccounts(t *testing.T) { + provider, mock, err := db_provider.NewDbProviderMock() + if err != nil { + t.Fatalf("Unexpected error stubbing DB: %v", err) + } + defer provider.Close() + + query := "SELECT id, config, 'account' AS dataType FROM accounts" + rows := sqlmock.NewRows([]string{"id", "data", "dataType"}). + AddRow("acc-1", `{"id":"acc-1"}`, "account"). + AddRow("acc-2", `{"id":"acc-2"}`, "account"). + AddRow("acc-null", nil, "account") // null data is skipped + mock.ExpectQuery(fmt.Sprintf("^%s$", regexp.QuoteMeta(query))).WillReturnRows(rows) + + fetcher := &dbFetcher{provider: provider, accountsQuery: query} + accounts, errs := fetcher.FetchAllAccounts(context.Background()) + + assertMockExpectations(t, mock) + assertErrorCount(t, 0, errs) + assertMapLength(t, 2, accounts) + assertHasData(t, accounts, "acc-1", `{"id":"acc-1"}`) + assertHasData(t, accounts, "acc-2", `{"id":"acc-2"}`) +} + // TestGoodResponse makes sure we interpret DB responses properly when all the stored requests are there. func TestGoodResponse(t *testing.T) { mockQuery := "SELECT id, data, 'request' AS dataType FROM req_table WHERE id IN (?) UNION ALL SELECT id, data, 'imp' as dataType FROM imp_table WHERE id IN (?, ?)" diff --git a/stored_requests/backends/empty_fetcher/fetcher.go b/stored_requests/backends/empty_fetcher/fetcher.go index 79de895ab60..762ed00ae1b 100644 --- a/stored_requests/backends/empty_fetcher/fetcher.go +++ b/stored_requests/backends/empty_fetcher/fetcher.go @@ -36,6 +36,11 @@ func (fetcher EmptyFetcher) FetchAccount(ctx context.Context, accountDefaultJSON return nil, []error{stored_requests.NotFoundError{ID: accountID, DataType: "Account"}} } +// FetchAllAccounts returns no accounts: the empty fetcher has no data to enumerate. +func (fetcher EmptyFetcher) FetchAllAccounts(ctx context.Context) (map[string]json.RawMessage, []error) { + return map[string]json.RawMessage{}, nil +} + func (fetcher EmptyFetcher) FetchCategories(ctx context.Context, primaryAdServer, publisherId, iabCategory string) (string, error) { return "", nil } diff --git a/stored_requests/backends/file_fetcher/fetcher.go b/stored_requests/backends/file_fetcher/fetcher.go index 615135f2db6..8ed800a3faa 100644 --- a/stored_requests/backends/file_fetcher/fetcher.go +++ b/stored_requests/backends/file_fetcher/fetcher.go @@ -70,6 +70,20 @@ func (fetcher *eagerFetcher) FetchAccount(ctx context.Context, accountDefaultsJS return completeJSON, nil } +// FetchAllAccounts returns every account held in memory, keyed by account ID. The +// bytes are raw (not defaults-merged); callers merge account defaults as needed. +func (fetcher *eagerFetcher) FetchAllAccounts(ctx context.Context) (map[string]json.RawMessage, []error) { + accountsDir, found := fetcher.FileSystem.Directories["accounts"] + if !found { + return map[string]json.RawMessage{}, nil + } + out := make(map[string]json.RawMessage, len(accountsDir.Files)) + for id, raw := range accountsDir.Files { + out[id] = raw + } + return out, nil +} + func (fetcher *eagerFetcher) FetchCategories(ctx context.Context, primaryAdServer, publisherId, iabCategory string) (string, error) { fileName := primaryAdServer diff --git a/stored_requests/backends/file_fetcher/fetcher_test.go b/stored_requests/backends/file_fetcher/fetcher_test.go index 8ca38aa873c..f8895635e09 100644 --- a/stored_requests/backends/file_fetcher/fetcher_test.go +++ b/stored_requests/backends/file_fetcher/fetcher_test.go @@ -99,6 +99,19 @@ func TestAccountFetcher(t *testing.T) { } +func TestFileFetcherFetchAllAccounts(t *testing.T) { + fetcher, err := NewFileFetcher("./test") + assert.NoError(t, err, "Failed to create test fetcher") + + bulk, ok := fetcher.(stored_requests.AllAccountsFetcher) + assert.True(t, ok, "file fetcher should support bulk account loading") + + accounts, errs := bulk.FetchAllAccounts(context.Background()) + assertErrorCount(t, 0, errs) + assert.Contains(t, accounts, "valid", "FetchAllAccounts should return every in-memory account") + assert.True(t, json.Valid(accounts["valid"]), "returned account bytes should be valid JSON") +} + func TestInvalidDirectory(t *testing.T) { _, err := NewFileFetcher("./nonexistant-directory") if err == nil { diff --git a/stored_requests/backends/http_fetcher/fetcher.go b/stored_requests/backends/http_fetcher/fetcher.go index a0e6daba5a2..0a295b6f87c 100644 --- a/stored_requests/backends/http_fetcher/fetcher.go +++ b/stored_requests/backends/http_fetcher/fetcher.go @@ -149,6 +149,16 @@ func (fetcher *HttpFetcher) FetchAccounts(ctx context.Context, accountIDs []stri fmt.Errorf(`Error fetching accounts %v via http: error reading response: %v`, accountIDs, err), } } + if httpResp.StatusCode == http.StatusNotFound { + errs := make([]error, 0, len(accountIDs)) + for _, accountID := range accountIDs { + errs = append(errs, stored_requests.NotFoundError{ + ID: accountID, + DataType: "Account", + }) + } + return nil, errs + } if httpResp.StatusCode != http.StatusOK { return nil, []error{ fmt.Errorf(`Error fetching accounts %v via http: unexpected response status %d`, accountIDs, httpResp.StatusCode), @@ -187,6 +197,15 @@ func (fetcher *HttpFetcher) FetchAccount(ctx context.Context, accountDefaultsJSO return completeJSON, nil } +// FetchAllAccounts is a no-op for the HTTP fetcher. Its endpoint contract is by-id +// only (GET ?account-ids=[...]) with no way to enumerate every account, so it +// contributes nothing to bulk cache preloading and reports no error. Accounts still +// load lazily on demand via FetchAccount. +func (fetcher *HttpFetcher) FetchAllAccounts(ctx context.Context) (map[string]json.RawMessage, []error) { + logger.Warnf("http_fetcher: bulk account preload is not supported by the by-id HTTP endpoint; accounts will load lazily on demand") + return map[string]json.RawMessage{}, nil +} + func (fetcher *HttpFetcher) FetchCategories(ctx context.Context, primaryAdServer, publisherId, iabCategory string) (string, error) { if fetcher.Categories == nil { fetcher.Categories = make(map[string]map[string]stored_requests.Category) diff --git a/stored_requests/backends/http_fetcher/fetcher_test.go b/stored_requests/backends/http_fetcher/fetcher_test.go index e98032480cd..9910bb3fe5b 100644 --- a/stored_requests/backends/http_fetcher/fetcher_test.go +++ b/stored_requests/backends/http_fetcher/fetcher_test.go @@ -10,6 +10,7 @@ import ( "testing" "time" + "github.com/prebid/prebid-server/v4/stored_requests" "github.com/prebid/prebid-server/v4/util/jsonutil" "github.com/stretchr/testify/assert" ) @@ -143,6 +144,15 @@ func TestFetchAccountsRfcCompliant(t *testing.T) { assertMapKeys(t, accData, "acc-1", "acc-2") } +func TestFetchAllAccountsNoop(t *testing.T) { + fetcher, close := newTestAccountFetcher(t, nil, false) + defer close() + + accounts, errs := fetcher.FetchAllAccounts(context.Background()) + assert.Empty(t, errs, "HTTP fetcher bulk load should be a no-op, not an error") + assert.Empty(t, accounts, "HTTP fetcher contributes no accounts to bulk preload") +} + func TestFetchAccounts(t *testing.T) { fetcher, close := newTestAccountFetcher(t, []string{"acc-1", "acc-2"}, false) defer close() @@ -170,6 +180,23 @@ func TestFetchAccountsBadJSON(t *testing.T) { assert.Nil(t, accData, "Fetching account with broken json should return nil account map") } +func TestFetchAccountsHTTP404ReturnsNotFoundErrors(t *testing.T) { + handler := func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNotFound) + } + server := httptest.NewServer(http.HandlerFunc(handler)) + defer server.Close() + fetcher := NewFetcher(server.Client(), server.URL, true) + + accData, errs := fetcher.FetchAccounts(context.Background(), []string{"acc-1", "acc-2"}) + + assert.Nil(t, accData, "HTTP 404 should not return account data") + assert.Equal(t, []error{ + stored_requests.NotFoundError{ID: "acc-1", DataType: "Account"}, + stored_requests.NotFoundError{ID: "acc-2", DataType: "Account"}, + }, errs) +} + func TestFetchAccountsNoIDsProvidedRfcCompliant(t *testing.T) { fetcher, close := newTestAccountFetcher(t, []string{"acc-1", "acc-2"}, true) defer close() diff --git a/stored_requests/config/config.go b/stored_requests/config/config.go index 87802589ebb..680f40fe27f 100644 --- a/stored_requests/config/config.go +++ b/stored_requests/config/config.go @@ -6,6 +6,7 @@ import ( "time" "github.com/julienschmidt/httprouter" + "github.com/prebid/prebid-server/v4/account" "github.com/prebid/prebid-server/v4/config" "github.com/prebid/prebid-server/v4/logger" "github.com/prebid/prebid-server/v4/metrics" @@ -35,7 +36,14 @@ import ( // As a side-effect, it will add some endpoints to the router if the config calls for it. // In the future we should look for ways to simplify this so that it's not doing two things. func CreateStoredRequests(cfg *config.StoredRequests, metricsEngine metrics.MetricsEngine, client *http.Client, router *httprouter.Router, provider db_provider.DbProvider) (fetcher stored_requests.AllFetcher, shutdown func()) { - // Create database connection if given options for one + return createLegacyCachedStoredRequests(cfg, metricsEngine, client, router, provider) +} + +func createStoredRequestSource(cfg *config.StoredRequests, client *http.Client, provider db_provider.DbProvider) stored_requests.AllFetcher { + return newFetcher(cfg, client, provider) +} + +func prepareStoredRequestsProvider(cfg *config.StoredRequests, provider db_provider.DbProvider) db_provider.DbProvider { if cfg.Database.ConnectionInfo.Database != "" { if provider == nil { logger.Infof("Connecting to Database for Stored %s. Driver=%s, DB=%s, host=%s, port=%d, user=%s", @@ -53,9 +61,29 @@ func CreateStoredRequests(cfg *config.StoredRequests, metricsEngine metrics.Metr logger.Fatalf("Multiple database connection settings found in config, only a single database connection is currently supported.") } } + return provider +} + +func createRawStoredRequests(cfg *config.StoredRequests, client *http.Client, provider db_provider.DbProvider) (fetcher stored_requests.AllFetcher, shutdown func()) { + provider = prepareStoredRequestsProvider(cfg, provider) + fetcher = createStoredRequestSource(cfg, client, provider) + shutdown = func() { + if provider == nil { + return + } + + if err := provider.Close(); err != nil { + logger.Errorf("Error closing DB connection: %v", err) + } + } + return +} + +func createLegacyCachedStoredRequests(cfg *config.StoredRequests, metricsEngine metrics.MetricsEngine, client *http.Client, router *httprouter.Router, provider db_provider.DbProvider) (fetcher stored_requests.AllFetcher, shutdown func()) { + provider = prepareStoredRequestsProvider(cfg, provider) eventProducers := newEventProducers(cfg, client, provider, metricsEngine, router) - fetcher = newFetcher(cfg, client, provider) + fetcher = createStoredRequestSource(cfg, client, provider) var shutdown1 func() @@ -110,7 +138,13 @@ func NewStoredRequests(cfg *config.Configuration, metricsEngine metrics.MetricsE fetcher2, shutdown2 := CreateStoredRequests(&cfg.StoredRequestsAMP, metricsEngine, client, router, provider) fetcher3, shutdown3 := CreateStoredRequests(&cfg.CategoryMapping, metricsEngine, client, router, provider) fetcher4, shutdown4 := CreateStoredRequests(&cfg.StoredVideo, metricsEngine, client, router, provider) - fetcher5, shutdown5 := CreateStoredRequests(&cfg.Accounts, metricsEngine, client, router, provider) + var fetcher5 stored_requests.AllFetcher + var shutdown5 func() + if cfg.Accounts.V2Enabled { + fetcher5, shutdown5 = createRawStoredRequests(&cfg.Accounts, client, provider) + } else { + fetcher5, shutdown5 = CreateStoredRequests(&cfg.Accounts, metricsEngine, client, router, provider) + } fetcher6, shutdown6 := CreateStoredRequests(&cfg.StoredResponses, metricsEngine, client, router, provider) fetcher = fetcher1.(stored_requests.Fetcher) @@ -120,6 +154,17 @@ func NewStoredRequests(cfg *config.Configuration, metricsEngine metrics.MetricsE accountsFetcher = fetcher5.(stored_requests.AccountFetcher) storedRespFetcher = fetcher6.(stored_requests.Fetcher) + // Fetchers 2.0: when enabled, wrap the raw account source with the typed + // cachekit fetcher. With v2_enabled=false the legacy byte-cache path is used + // unchanged. + if cfg.Accounts.V2Enabled { + v2Accounts, err := account.NewCacheKitAccountFetcher(fetcher5, cfg.Accounts.CacheV2, cfg.AccountDefaultsJSON(), nil, metricsEngine) + if err != nil { + logger.Fatalf("Failed to initialize Fetchers 2.0 account fetcher: %v", err) + } + accountsFetcher = v2Accounts + } + shutdown = func() { shutdown1() shutdown2() @@ -157,8 +202,9 @@ func newFetcher(cfg *config.StoredRequests, client *http.Client, provider db_pro } if cfg.Database.FetcherQueries.QueryTemplate != "" { logger.Infof("Loading Stored %s data via Database.\nQuery: %s", cfg.DataType(), cfg.Database.FetcherQueries.QueryTemplate) - idList = append(idList, db_fetcher.NewFetcher(provider, - cfg.Database.FetcherQueries.QueryTemplate, cfg.Database.FetcherQueries.QueryTemplate)) + idList = append(idList, db_fetcher.NewFetcherWithAccountsQuery(provider, + cfg.Database.FetcherQueries.QueryTemplate, cfg.Database.FetcherQueries.QueryTemplate, + cfg.Database.CacheInitialization.Query)) } else if cfg.Database.CacheInitialization.Query != "" && cfg.Database.PollUpdates.Query != "" { //in this case data will be loaded to cache via poll for updates event idList = append(idList, empty_fetcher.EmptyFetcher{}) diff --git a/stored_requests/config/config_test.go b/stored_requests/config/config_test.go index 913f8c3959b..12ca5fde40f 100644 --- a/stored_requests/config/config_test.go +++ b/stored_requests/config/config_test.go @@ -4,17 +4,21 @@ import ( "context" "encoding/json" "errors" + "fmt" "net/http" "net/http/httptest" "regexp" + "sync/atomic" "testing" "github.com/stretchr/testify/assert" sqlmock "github.com/DATA-DOG/go-sqlmock" "github.com/julienschmidt/httprouter" + accountservice "github.com/prebid/prebid-server/v4/account" "github.com/prebid/prebid-server/v4/config" "github.com/prebid/prebid-server/v4/metrics" + metricsconfig "github.com/prebid/prebid-server/v4/metrics/config" "github.com/prebid/prebid-server/v4/stored_requests" "github.com/prebid/prebid-server/v4/stored_requests/backends/db_provider" "github.com/prebid/prebid-server/v4/stored_requests/backends/empty_fetcher" @@ -163,6 +167,47 @@ func TestNewHTTPEvents(t *testing.T) { assertHttpWithURL(t, evProducers[0], server1.URL) } +func TestNewStoredRequestsV2AccountsSkipsLegacyAccountCache(t *testing.T) { + var calls int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"accounts":{"pub-1":{"id":"pub-1"}}}`) + })) + defer server.Close() + + cfg := &config.Configuration{ + Accounts: config.StoredRequests{ + HTTP: config.HTTPFetcherConfig{ + Endpoint: server.URL, + UseRfcCompliantBuilder: true, + }, + InMemoryCache: config.InMemoryCache{ + Type: "unbounded", + }, + V2Enabled: true, + CacheV2: config.CacheKitConfig{ + Type: "none", + }, + }, + } + cfg.Accounts.SetDataType(config.AccountDataType) + assert.NoError(t, cfg.MarshalAccountDefaults()) + + shutdown, _, _, accountsFetcher, _, _, _ := NewStoredRequests(cfg, &metricsconfig.NilMetricsEngine{}, server.Client(), httprouter.New()) + defer shutdown() + + account, errs := accountservice.GetAccount(context.Background(), cfg, accountsFetcher, "pub-1", &metricsconfig.NilMetricsEngine{}) + assert.Empty(t, errs) + assert.Equal(t, "pub-1", account.ID) + + account, errs = accountservice.GetAccount(context.Background(), cfg, accountsFetcher, "pub-1", &metricsconfig.NilMetricsEngine{}) + assert.Empty(t, errs) + assert.Equal(t, "pub-1", account.ID) + + assert.Equal(t, int32(2), atomic.LoadInt32(&calls), "v2 cache.type=none should reach the HTTP source every time, even if the legacy account cache is configured") +} + func TestNewEmptyCache(t *testing.T) { cache := newCache(&config.StoredRequests{InMemoryCache: config.InMemoryCache{Type: "none"}}) assert.True(t, isEmptyCacheType(cache.Requests), "The newCache method should return an empty Request cache") diff --git a/stored_requests/fetcher.go b/stored_requests/fetcher.go index 2abfb330529..c510d988313 100644 --- a/stored_requests/fetcher.go +++ b/stored_requests/fetcher.go @@ -3,6 +3,7 @@ package stored_requests import ( "context" "encoding/json" + "errors" "fmt" "github.com/prebid/prebid-server/v4/metrics" @@ -31,6 +32,15 @@ type AccountFetcher interface { FetchAccount(ctx context.Context, accountDefaultJSON json.RawMessage, accountID string) (json.RawMessage, []error) } +// AllAccountsFetcher is an optional capability a fetcher may implement to return +// every account in a single call, enabling bulk cache preloading. Sources that +// cannot enumerate all accounts (e.g. a by-id HTTP endpoint) do not implement it, +// or return an error indicating the capability is unavailable. The returned bytes +// are raw account JSON (not defaults-merged); callers merge defaults as needed. +type AllAccountsFetcher interface { + FetchAllAccounts(ctx context.Context) (map[string]json.RawMessage, []error) +} + type CategoryFetcher interface { // FetchCategories fetches the ad-server/publisher specific category for the given IAB category FetchCategories(ctx context.Context, primaryAdServer, publisherId, iabCategory string) (string, error) @@ -230,6 +240,14 @@ func (f *fetcherWithCache) FetchCategories(ctx context.Context, primaryAdServer, return "", nil } +// FetchAllAccounts delegates to the wrapped fetcher when it supports bulk account +// enumeration; otherwise it reports the capability as unavailable. +func (f *fetcherWithCache) FetchAllAccounts(ctx context.Context) (map[string]json.RawMessage, []error) { + if bulk, ok := f.fetcher.(AllAccountsFetcher); ok { + return bulk.FetchAllAccounts(ctx) + } + return nil, []error{errors.New("stored_requests: underlying account fetcher does not support bulk loading")} +} func findLeftovers(ids []string, data map[string]json.RawMessage) (leftovers []string) { leftovers = make([]string, 0, len(ids)-len(data)) for _, id := range ids { diff --git a/stored_requests/multifetcher.go b/stored_requests/multifetcher.go index f301600d358..130b0109c7a 100644 --- a/stored_requests/multifetcher.go +++ b/stored_requests/multifetcher.go @@ -70,6 +70,32 @@ func (mf MultiFetcher) FetchCategories(ctx context.Context, primaryAdServer, pub return "", NotFoundError{errtype, "Category"} } +// FetchAllAccounts merges the bulk account sets from every sub-fetcher that +// supports enumeration. Earlier fetchers win on ID conflicts (first-wins, +// mirroring FetchAccount). Sub-fetchers that do not support bulk loading are +// skipped. If no sub-fetcher supports it, an empty map is returned. +func (mf MultiFetcher) FetchAllAccounts(ctx context.Context) (map[string]json.RawMessage, []error) { + out := make(map[string]json.RawMessage) + var errs []error + for _, f := range mf { + bulk, ok := f.(AllAccountsFetcher) + if !ok { + continue + } + accounts, ferrs := bulk.FetchAllAccounts(ctx) + if len(ferrs) > 0 { + errs = append(errs, ferrs...) + continue + } + for id, raw := range accounts { + if _, exists := out[id]; !exists { + out[id] = raw + } + } + } + return out, errs +} + func addAll(base map[string]json.RawMessage, toAdd map[string]json.RawMessage) { for k, v := range toAdd { base[k] = v diff --git a/stored_requests/multifetcher_test.go b/stored_requests/multifetcher_test.go index beca325e37a..7d81a987d93 100644 --- a/stored_requests/multifetcher_test.go +++ b/stored_requests/multifetcher_test.go @@ -176,3 +176,62 @@ func TestMultiFetcherAccountNotFound(t *testing.T) { assert.Nil(t, account) assert.EqualError(t, errs[0], NotFoundError{"MISSING", "Account"}.Error()) } + +// noAccountsFetcher implements AllFetcher but not AllAccountsFetcher, so MultiFetcher +// should skip it during bulk enumeration. +type noAccountsFetcher struct{} + +func (noAccountsFetcher) FetchRequests(ctx context.Context, requestIDs []string, impIDs []string) (map[string]json.RawMessage, map[string]json.RawMessage, []error) { + return nil, nil, nil +} +func (noAccountsFetcher) FetchResponses(ctx context.Context, ids []string) (map[string]json.RawMessage, []error) { + return nil, nil +} +func (noAccountsFetcher) FetchAccount(ctx context.Context, def json.RawMessage, id string) (json.RawMessage, []error) { + return nil, []error{NotFoundError{id, "Account"}} +} +func (noAccountsFetcher) FetchCategories(ctx context.Context, primaryAdServer, publisherId, iabCategory string) (string, error) { + return "", nil +} + +// bulkAccountsFetcher additionally supports bulk account enumeration. +type bulkAccountsFetcher struct { + noAccountsFetcher + accounts map[string]json.RawMessage + errs []error +} + +func (b bulkAccountsFetcher) FetchAllAccounts(ctx context.Context) (map[string]json.RawMessage, []error) { + return b.accounts, b.errs +} + +func TestMultiFetcherFetchAllAccounts(t *testing.T) { + first := bulkAccountsFetcher{accounts: map[string]json.RawMessage{ + "a": json.RawMessage(`{"src":"first"}`), + "b": json.RawMessage(`{"src":"first"}`), + }} + second := bulkAccountsFetcher{accounts: map[string]json.RawMessage{ + "b": json.RawMessage(`{"src":"second"}`), // conflicts with first; first wins + "c": json.RawMessage(`{"src":"second"}`), + }} + fetcher := &MultiFetcher{first, noAccountsFetcher{}, second} + + accounts, errs := fetcher.FetchAllAccounts(context.Background()) + + assert.Empty(t, errs) + assert.Len(t, accounts, 3, "should merge accounts from all bulk-capable fetchers") + assert.JSONEq(t, `{"src":"first"}`, string(accounts["a"])) + assert.JSONEq(t, `{"src":"first"}`, string(accounts["b"]), "earlier fetcher wins on ID conflict") + assert.JSONEq(t, `{"src":"second"}`, string(accounts["c"])) +} + +func TestMultiFetcherFetchAllAccountsPropagatesErrors(t *testing.T) { + failing := bulkAccountsFetcher{errs: []error{errors.New("db down")}} + ok := bulkAccountsFetcher{accounts: map[string]json.RawMessage{"a": json.RawMessage(`{}`)}} + fetcher := &MultiFetcher{failing, ok} + + accounts, errs := fetcher.FetchAllAccounts(context.Background()) + + assert.Len(t, errs, 1, "a failing fetcher's error should be surfaced") + assert.Contains(t, accounts, "a", "healthy fetchers still contribute their accounts") +}