diff --git a/targeting/expbench/bench.go b/targeting/expbench/bench.go new file mode 100644 index 00000000..66f2ec1d --- /dev/null +++ b/targeting/expbench/bench.go @@ -0,0 +1,279 @@ +package expbench + +import ( + "context" + "fmt" + "math/rand/v2" + "sort" + "time" +) + +// LoadProfile describes a synthetic user's exposure log shape. +type LoadProfile struct { + Name string + NumImpressions int // total impressions over Window + Window time.Duration // spread evenly over this duration ending at "now" + KeysPerImpression int // how many fcap_keys each impression carries +} + +// GenerateLog produces a deterministic synthetic exposure log for the +// profile, ending at endTimestamp (unix seconds). Impressions are spread +// uniformly across the window. Each impression's fcap_keys are drawn from +// a pool large enough to make the keys diverse but small enough to make +// them re-occur (so frequency caps actually fire in the bench). +func GenerateLog(profile LoadProfile, endTimestamp int64, seed uint64) []Impression { + r := rand.New(rand.NewPCG(seed, seed^0xdeadbeef)) + imps := make([]Impression, profile.NumImpressions) + windowSec := int64(profile.Window.Seconds()) + startTS := endTimestamp - windowSec + + // Pool of distinct fcap_keys spanning ~3 dimensions. + const numCampaigns = 50 + const numAdvertisers = 20 + const numCreatives = 200 + const numLineItems = 100 + + for i := range imps { + // Spread timestamps roughly uniformly over the window with a small + // jitter, then sort. + ts := startTS + int64(float64(windowSec)*float64(i)/float64(profile.NumImpressions)) + ts += int64(r.IntN(60)) // up to 1 minute jitter + + keys := make([]string, 0, profile.KeysPerImpression) + // Always include a campaign and advertiser; fill the rest with creatives/line-items. + keys = append(keys, fmt.Sprintf("campaign:%d", r.IntN(numCampaigns))) + if profile.KeysPerImpression >= 2 { + keys = append(keys, fmt.Sprintf("advertiser:%d", r.IntN(numAdvertisers))) + } + if profile.KeysPerImpression >= 3 { + keys = append(keys, fmt.Sprintf("creative:%d", r.IntN(numCreatives))) + } + if profile.KeysPerImpression >= 4 { + keys = append(keys, fmt.Sprintf("lineitem:%d", r.IntN(numLineItems))) + } + for j := 4; j < profile.KeysPerImpression; j++ { + keys = append(keys, fmt.Sprintf("dim%d:%d", j, r.IntN(50))) + } + + imps[i] = Impression{ + ImpressionID: fmt.Sprintf("imp-%d-%d", seed, i), + Timestamp: ts, + FcapKeys: keys, + } + } + sort.Slice(imps, func(i, j int) bool { return imps[i].Timestamp < imps[j].Timestamp }) + return imps +} + +// Stats is per-operation latency in microseconds. +type Stats struct { + N int + P50us float64 + P95us float64 + P99us float64 + MeanUs float64 +} + +// Result is the bench output for one (variant, profile, rule-window) cell. +type Result struct { + Variant string + Profile string + RuleWindow string + + WriteSteady Stats // write into a full 30-day log + ReadSingle Stats // single fcap_key eligibility + ReadBatch100 Stats // 100 fcap_keys at once + ReadBatch1000 Stats // 1000 fcap_keys at once + + CleanupUs float64 // single cleanup call, drops one day's worth + CleanupBytes int64 // memory after cleanup (decreased) + + MemoryBytes int64 // valkey MEMORY USAGE after seeding +} + +// computeStats returns p50/p95/p99/mean over a slice of latencies in +// microseconds. +func computeStats(latenciesUs []float64) Stats { + if len(latenciesUs) == 0 { + return Stats{} + } + sorted := make([]float64, len(latenciesUs)) + copy(sorted, latenciesUs) + sort.Float64s(sorted) + pct := func(p float64) float64 { + idx := int(float64(len(sorted)-1) * p) + return sorted[idx] + } + var sum float64 + for _, v := range latenciesUs { + sum += v + } + return Stats{ + N: len(latenciesUs), + P50us: pct(0.50), + P95us: pct(0.95), + P99us: pct(0.99), + MeanUs: sum / float64(len(latenciesUs)), + } +} + +// FormatResultsTable renders results as a markdown table. +func FormatResultsTable(results []Result) string { + out := "| variant | profile | rule | write p50 | write p95 | read1 p50 | read100 p50 | read1k p50 | cleanup | memory |\n" + out += "|---|---|---|---:|---:|---:|---:|---:|---:|---:|\n" + for _, r := range results { + out += fmt.Sprintf("| %s | %s | %s | %.0f µs | %.0f µs | %.0f µs | %.0f µs | %.0f µs | %.0f µs | %s |\n", + r.Variant, r.Profile, r.RuleWindow, + r.WriteSteady.P50us, r.WriteSteady.P95us, + r.ReadSingle.P50us, + r.ReadBatch100.P50us, r.ReadBatch1000.P50us, + r.CleanupUs, + fmtBytes(r.MemoryBytes), + ) + } + return out +} + +func fmtBytes(b int64) string { + switch { + case b >= 1<<20: + return fmt.Sprintf("%.1f MB", float64(b)/(1<<20)) + case b >= 1<<10: + return fmt.Sprintf("%.1f KB", float64(b)/(1<<10)) + default: + return fmt.Sprintf("%d B", b) + } +} + +// RunBench runs the full benchmark suite for one (variant, profile, +// rule-window) cell. The rule's window controls the server-side range +// filter on ZSET reads — a 24h rule lets ZSETs skip 96% of a 30-day +// log; a 30d rule pulls everything. +func RunBench(ctx context.Context, v Variant, profile LoadProfile, rules []FrequencyRule, ruleLabel, userID string, now int64, iterations, batchPoolSize int) (Result, error) { + res := Result{Variant: v.Name(), Profile: profile.Name, RuleWindow: ruleLabel} + + // Reset to a clean slate. + if err := v.Reset(ctx, userID); err != nil { + return res, fmt.Errorf("reset: %w", err) + } + imps := GenerateLog(profile, now, 0xC0FFEE) + if err := v.Seed(ctx, userID, imps); err != nil { + return res, fmt.Errorf("seed: %w", err) + } + mem, err := v.MemoryUsage(ctx, userID) + if err != nil { + return res, fmt.Errorf("memory usage: %w", err) + } + res.MemoryBytes = mem + + // Pre-build a pool of distinct fcap_keys to drive read benchmarks. Reuse + // keys that actually appear in the seeded log so reads find matches. + keyPool := make([]string, 0, batchPoolSize) + keySeen := make(map[string]struct{}, batchPoolSize) + for _, imp := range imps { + for _, k := range imp.FcapKeys { + if _, seen := keySeen[k]; seen { + continue + } + keySeen[k] = struct{}{} + keyPool = append(keyPool, k) + if len(keyPool) >= batchPoolSize { + break + } + } + if len(keyPool) >= batchPoolSize { + break + } + } + + // Steady-state write: append to the already-full log. + writeLatencies := make([]float64, 0, iterations) + for i := 0; i < iterations; i++ { + imp := Impression{ + ImpressionID: fmt.Sprintf("write-%d", i), + Timestamp: now + int64(i), + FcapKeys: []string{"campaign:1", "advertiser:2", "creative:3", "lineitem:4"}, + } + start := time.Now() + if err := v.Write(ctx, userID, imp); err != nil { + return res, fmt.Errorf("write: %w", err) + } + writeLatencies = append(writeLatencies, float64(time.Since(start).Microseconds())) + } + res.WriteSteady = computeStats(writeLatencies) + + // Single-rule read. + singleLatencies := make([]float64, 0, iterations) + for i := 0; i < iterations; i++ { + k := keyPool[i%len(keyPool)] + kh := HashKey(k) + start := time.Now() + _, _, err := v.ReadAndCheck(ctx, userID, kh, rules, now) + if err != nil { + return res, fmt.Errorf("read single: %w", err) + } + singleLatencies = append(singleLatencies, float64(time.Since(start).Microseconds())) + } + res.ReadSingle = computeStats(singleLatencies) + + // Batch read: 100 fcap_keys. + batch100Latencies := make([]float64, 0, iterations) + keys100 := make([]uint64, 0, 100) + for i := 0; i < 100 && i < len(keyPool); i++ { + keys100 = append(keys100, HashKey(keyPool[i])) + } + for i := 0; i < iterations; i++ { + start := time.Now() + _, err := v.ReadBatchCheck(ctx, userID, keys100, rules, now) + if err != nil { + return res, fmt.Errorf("read batch 100: %w", err) + } + batch100Latencies = append(batch100Latencies, float64(time.Since(start).Microseconds())) + } + res.ReadBatch100 = computeStats(batch100Latencies) + + // Batch read: 1000 fcap_keys (synthesizing keys beyond the pool with + // fresh hashes that won't match any entry — that mirrors the real + // case where many candidate packages aren't ones the user has seen). + batch1000Latencies := make([]float64, 0, iterations) + keys1000 := make([]uint64, 0, 1000) + for i := 0; i < 1000; i++ { + if i < len(keyPool) { + keys1000 = append(keys1000, HashKey(keyPool[i])) + } else { + keys1000 = append(keys1000, HashKey(fmt.Sprintf("nomatch:%d", i))) + } + } + for i := 0; i < iterations/2; i++ { // half iterations: each call is heavier + start := time.Now() + _, err := v.ReadBatchCheck(ctx, userID, keys1000, rules, now) + if err != nil { + return res, fmt.Errorf("read batch 1000: %w", err) + } + batch1000Latencies = append(batch1000Latencies, float64(time.Since(start).Microseconds())) + } + res.ReadBatch1000 = computeStats(batch1000Latencies) + + // Cleanup: drop entries older than (now - 29 days). Simulates the + // hourly cleanup pass that drops entries falling out of the 30-day + // window. Re-seed first so per-write writes haven't already + // rebalanced the log. + if err := v.Reset(ctx, userID); err != nil { + return res, fmt.Errorf("reset before cleanup: %w", err) + } + if err := v.Seed(ctx, userID, imps); err != nil { + return res, fmt.Errorf("re-seed for cleanup: %w", err) + } + cleanupStart := time.Now() + if err := v.Cleanup(ctx, userID, now, 29*24*time.Hour); err != nil { + return res, fmt.Errorf("cleanup: %w", err) + } + res.CleanupUs = float64(time.Since(cleanupStart).Microseconds()) + memAfter, err := v.MemoryUsage(ctx, userID) + if err != nil { + return res, fmt.Errorf("memory after cleanup: %w", err) + } + res.CleanupBytes = memAfter + + return res, nil +} diff --git a/targeting/expbench/bench_split_test.go b/targeting/expbench/bench_split_test.go new file mode 100644 index 00000000..ceb82a02 --- /dev/null +++ b/targeting/expbench/bench_split_test.go @@ -0,0 +1,180 @@ +package expbench + +import ( + "context" + "fmt" + "os" + "runtime" + "sort" + "testing" + "time" + + "github.com/redis/go-redis/v9" +) + +// TestBenchSplit measures the read path with three timers per call: +// total = fetch + process. Fetch is "wait on valkey + deserialize wire +// protocol into Go types"; process is "in-memory eligibility scan with +// no I/O." Goal: see where the cost actually lives, not just the +// end-to-end number. +// +// VALKEY_ADDR=localhost:6380 go test -run TestBenchSplit -v -timeout 5m ./targeting/expbench/ +func TestBenchSplit(t *testing.T) { + addr := os.Getenv("VALKEY_ADDR") + if addr == "" { + addr = "localhost:6380" + } + rdb := redis.NewClient(&redis.Options{Addr: addr}) + ctx, cancel := context.WithTimeout(context.Background(), 4*time.Minute) + defer cancel() + if err := rdb.Ping(ctx).Err(); err != nil { + t.Skipf("valkey unreachable at %s: %v", addr, err) + } + + variants := []SplitVariant{ + NewBinaryStore(rdb), + NewZSetArrayStore(rdb), + NewZSetPerKeyStore(rdb), + NewZSetPerKeyedStore(rdb), + } + profiles := []LoadProfile{ + {Name: "median 3K", NumImpressions: 3_000, Window: 30 * 24 * time.Hour, KeysPerImpression: 4}, + {Name: "heavy 30K", NumImpressions: 30_000, Window: 30 * 24 * time.Hour, KeysPerImpression: 4}, + } + + now := time.Now().Unix() + const iterations = 200 + ruleSets := []struct { + name string + rules []FrequencyRule + }{ + {"24h", []FrequencyRule{{MaxCount: 100, Window: 24 * time.Hour}}}, + {"30d", []FrequencyRule{{MaxCount: 100, Window: 30 * 24 * time.Hour}}}, + } + + type cell struct { + variant string + profile string + ruleWindow string + batch int + fetchUs float64 + processUs float64 + totalUs float64 + fetchBytes int + fetchMembers int + processAllocs uint64 + } + var rows []cell + + for _, v := range variants { + for _, p := range profiles { + userID := "split-" + v.Name() + "-" + p.Name + _ = v.Reset(ctx, userID) + imps := GenerateLog(p, now, 0xC0FFEE) + if err := v.Seed(ctx, userID, imps); err != nil { + t.Fatalf("seed %s/%s: %v", v.Name(), p.Name, err) + } + + // Build the same fcap_key pool as the other bench. + seenKeys := make(map[string]struct{}) + pool := make([]string, 0, 1000) + for _, imp := range imps { + for _, k := range imp.FcapKeys { + if _, seen := seenKeys[k]; seen { + continue + } + seenKeys[k] = struct{}{} + pool = append(pool, k) + if len(pool) >= 1000 { + break + } + } + if len(pool) >= 1000 { + break + } + } + + for _, rs := range ruleSets { + fetchWindow := rs.rules[0].Window + for _, batchSize := range []int{1, 100, 1000} { + keys := make([]uint64, 0, batchSize) + for i := 0; i < batchSize; i++ { + if i < len(pool) { + keys = append(keys, HashKey(pool[i])) + } else { + keys = append(keys, HashKey(fmt.Sprintf("nomatch:%d", i))) + } + } + + iters := iterations + if batchSize == 1000 { + iters = iterations / 2 + } + + fetchLatencies := make([]float64, 0, iters) + processLatencies := make([]float64, 0, iters) + totalLatencies := make([]float64, 0, iters) + var fetchBytes, fetchMembers int + var totalProcessAllocs uint64 + for i := 0; i < iters; i++ { + tStart := time.Now() + raw, err := v.Fetch(ctx, userID, fetchWindow, now) + fetchEnd := time.Now() + if err != nil { + t.Fatalf("fetch %s/%s: %v", v.Name(), p.Name, err) + } + if i == 0 { + fetchBytes, fetchMembers = SizeOfFetch(raw) + } + + var ms1, ms2 runtime.MemStats + runtime.ReadMemStats(&ms1) + _ = v.Process(raw, keys, rs.rules, now) + processEnd := time.Now() + runtime.ReadMemStats(&ms2) + totalProcessAllocs += ms2.TotalAlloc - ms1.TotalAlloc + + fetchLatencies = append(fetchLatencies, float64(fetchEnd.Sub(tStart).Microseconds())) + processLatencies = append(processLatencies, float64(processEnd.Sub(fetchEnd).Microseconds())) + totalLatencies = append(totalLatencies, float64(processEnd.Sub(tStart).Microseconds())) + } + rows = append(rows, cell{ + variant: v.Name(), + profile: p.Name, + ruleWindow: rs.name, + batch: batchSize, + fetchUs: median(fetchLatencies), + processUs: median(processLatencies), + totalUs: median(totalLatencies), + fetchBytes: fetchBytes, + fetchMembers: fetchMembers, + processAllocs: totalProcessAllocs / uint64(iters), + }) + } + } + _ = v.Reset(ctx, userID) + } + } + + out := "\n| variant | profile | rule | batch | fetch p50 | process p50 | total p50 | fetch bytes | members | proc allocs |\n" + out += "|---|---|---|---:|---:|---:|---:|---:|---:|---:|\n" + for _, r := range rows { + out += fmt.Sprintf("| %s | %s | %s | %d | %.0f µs | %.0f µs | %.0f µs | %s | %d | %s |\n", + r.variant, r.profile, r.ruleWindow, r.batch, + r.fetchUs, r.processUs, r.totalUs, + fmtBytes(int64(r.fetchBytes)), r.fetchMembers, + fmtBytes(int64(r.processAllocs)), + ) + } + t.Log(out) +} + +func median(xs []float64) float64 { + if len(xs) == 0 { + return 0 + } + c := make([]float64, len(xs)) + copy(c, xs) + sort.Float64s(c) + return c[len(c)/2] +} diff --git a/targeting/expbench/bench_test.go b/targeting/expbench/bench_test.go new file mode 100644 index 00000000..495f52f0 --- /dev/null +++ b/targeting/expbench/bench_test.go @@ -0,0 +1,158 @@ +package expbench + +import ( + "context" + "os" + "testing" + "time" + + "github.com/redis/go-redis/v9" +) + +// TestBench is the entry point for the storage-shape comparison. +// Run against a live valkey/redis with VALKEY_ADDR=host:port (default +// localhost:6380, matching the docker container started for this work). +// +// VALKEY_ADDR=localhost:6380 go test -run TestBench -v -timeout 5m ./targeting/expbench/ +func TestBench(t *testing.T) { + addr := os.Getenv("VALKEY_ADDR") + if addr == "" { + addr = "localhost:6380" + } + rdb := redis.NewClient(&redis.Options{Addr: addr}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + defer cancel() + if err := rdb.Ping(ctx).Err(); err != nil { + t.Skipf("valkey unreachable at %s: %v", addr, err) + } + + variants := []Variant{ + NewBinaryStore(rdb), + NewZSetArrayStore(rdb), + NewZSetPerKeyStore(rdb), + NewZSetPerKeyedStore(rdb), + NewBucketDayStore(rdb), + NewBucketCountStore(rdb), + } + profiles := []LoadProfile{ + {Name: "median 3K", NumImpressions: 3_000, Window: 30 * 24 * time.Hour, KeysPerImpression: 4}, + {Name: "heavy 30K", NumImpressions: 30_000, Window: 30 * 24 * time.Hour, KeysPerImpression: 4}, + } + ruleSets := []struct { + label string + rules []FrequencyRule + }{ + {"24h", []FrequencyRule{{MaxCount: 100, Window: 24 * time.Hour}}}, + {"30d", []FrequencyRule{{MaxCount: 100, Window: 30 * 24 * time.Hour}}}, + } + + now := time.Now().Unix() + const iterations = 200 + const batchPool = 1000 + + results := make([]Result, 0, len(variants)*len(profiles)*len(ruleSets)) + for _, v := range variants { + for _, p := range profiles { + for _, rs := range ruleSets { + // Bucket-day is a singleton-per-day model — only meaningful + // against day-window rules. The 30d rule case doesn't fit + // its model; skip it. + // Bucket variants are singleton/count-per-day models — only + // meaningful against day-window rules. + if (v.Name() == "bucket-day" || v.Name() == "bucket-count") && rs.label != "24h" { + continue + } + userID := v.Name() + "-" + p.Name + "-" + rs.label + t.Logf("running %s / %s / %s", v.Name(), p.Name, rs.label) + r, err := RunBench(ctx, v, p, rs.rules, rs.label, userID, now, iterations, batchPool) + if err != nil { + t.Errorf("RunBench(%s, %s, %s): %v", v.Name(), p.Name, rs.label, err) + continue + } + results = append(results, r) + _ = v.Reset(ctx, userID) + } + } + } + + t.Log("\n" + FormatResultsTable(results)) +} + +// TestSanity_Equivalence verifies that all three variants produce the same +// eligibility answer for the same workload. If they disagree, the perf +// comparison is meaningless because they aren't computing the same thing. +func TestSanity_Equivalence(t *testing.T) { + addr := os.Getenv("VALKEY_ADDR") + if addr == "" { + addr = "localhost:6380" + } + rdb := redis.NewClient(&redis.Options{Addr: addr}) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := rdb.Ping(ctx).Err(); err != nil { + t.Skipf("valkey unreachable at %s: %v", addr, err) + } + + now := time.Now().Unix() + imps := []Impression{ + {ImpressionID: "imp-1", Timestamp: now - 3600, FcapKeys: []string{"campaign:1", "advertiser:7"}}, + {ImpressionID: "imp-2", Timestamp: now - 1800, FcapKeys: []string{"campaign:1", "advertiser:7"}}, + {ImpressionID: "imp-3", Timestamp: now - 900, FcapKeys: []string{"campaign:1", "creative:3"}}, + {ImpressionID: "imp-4", Timestamp: now - 100, FcapKeys: []string{"campaign:2"}}, + {ImpressionID: "imp-old", Timestamp: now - 90*86400, FcapKeys: []string{"campaign:1"}}, + } + rules := []FrequencyRule{{MaxCount: 3, Window: 24 * time.Hour}} + keys := []string{"campaign:1", "advertiser:7", "creative:3", "campaign:2", "campaign:99"} + + type answer struct { + capped map[string]bool + latest map[string]int64 + } + collect := func(v Variant) answer { + userID := "sanity-" + v.Name() + _ = v.Reset(ctx, userID) + if err := v.Seed(ctx, userID, imps); err != nil { + t.Fatalf("seed %s: %v", v.Name(), err) + } + out := answer{capped: map[string]bool{}, latest: map[string]int64{}} + for _, k := range keys { + capped, ts, err := v.ReadAndCheck(ctx, userID, HashKey(k), rules, now) + if err != nil { + t.Fatalf("read %s/%s: %v", v.Name(), k, err) + } + out.capped[k] = capped + out.latest[k] = ts + } + _ = v.Reset(ctx, userID) + return out + } + + a := collect(NewBinaryStore(rdb)) + b := collect(NewZSetArrayStore(rdb)) + c := collect(NewZSetPerKeyStore(rdb)) + d := collect(NewZSetPerKeyedStore(rdb)) + + for _, k := range keys { + if a.capped[k] != b.capped[k] || a.capped[k] != c.capped[k] || a.capped[k] != d.capped[k] { + t.Errorf("capped(%s): binary=%v zset-array=%v zset-perkey=%v zset-perkeyed=%v", + k, a.capped[k], b.capped[k], c.capped[k], d.capped[k]) + } + if a.latest[k] != b.latest[k] || a.latest[k] != c.latest[k] || a.latest[k] != d.latest[k] { + t.Errorf("latest(%s): binary=%v zset-array=%v zset-perkey=%v zset-perkeyed=%v", + k, a.latest[k], b.latest[k], c.latest[k], d.latest[k]) + } + } + + // campaign:1 has 3 in 24h window (imp-1, imp-2, imp-3) → MaxCount=3 → capped. + if !a.capped["campaign:1"] { + t.Errorf("expected campaign:1 to be capped (3 hits in 24h, MaxCount=3); binary said no") + } + // campaign:99 should never be capped. + if a.capped["campaign:99"] { + t.Errorf("campaign:99 should not be capped") + } + // imp-old (90 days ago) must not contribute. + if a.latest["campaign:1"] < now-3600 { + t.Errorf("latest(campaign:1) = %d should be from imp-1 (now-3600), not imp-old", a.latest["campaign:1"]) + } +} diff --git a/targeting/expbench/binary.go b/targeting/expbench/binary.go new file mode 100644 index 00000000..8f045f43 --- /dev/null +++ b/targeting/expbench/binary.go @@ -0,0 +1,275 @@ +package expbench + +import ( + "context" + "encoding/binary" + "errors" + "time" + + "github.com/redis/go-redis/v9" +) + +// Binary log entry layout (88 bytes, 8-byte aligned): +// +// timestamp(8) + impressionHash(8) + keyCount(1) + padding(7) + fcapKeyHash[0..7](8 each) +// +// Header (4 bytes): version(uint16) + entrySize(uint16). Stored as a single +// byte string in valkey under user:exposures:{uid}. +const ( + binHeaderSize = 4 + binEntrySize = 88 + binVersion = 1 + binTSOffset = 0 + binImpOffset = 8 + binCountOffset = 16 + binKeysOffset = 24 +) + +// BinaryStore implements the binary-log variant: read-modify-write of a +// single byte slab per user. Generalized from the current +// `targeting/exposure_binary.go` to carry up to MaxKeysPerImpression +// fcap_key hashes per entry instead of fixed package/campaign slots. +type BinaryStore struct { + rdb redis.Cmdable +} + +// NewBinaryStore returns a BinaryStore backed by the given redis client. +func NewBinaryStore(rdb redis.Cmdable) *BinaryStore { return &BinaryStore{rdb: rdb} } + +// Name is used in benchmark output. +func (s *BinaryStore) Name() string { return "binary" } + +// Key for a user's exposure log. +func binaryKey(userID string) string { return "user:exposures:bin:" + userID } + +func encodeEntry(buf []byte, imp Impression) { + if len(imp.FcapKeys) > MaxKeysPerImpression { + // Bench setup ensures K <= MaxKeysPerImpression; truncation here + // makes a deterministic choice if violated. + imp.FcapKeys = imp.FcapKeys[:MaxKeysPerImpression] + } + binary.LittleEndian.PutUint64(buf[binTSOffset:], uint64(imp.Timestamp)) //nolint:gosec + binary.LittleEndian.PutUint64(buf[binImpOffset:], HashKey(imp.ImpressionID)) + buf[binCountOffset] = byte(len(imp.FcapKeys)) + // padding bytes 17..23 left zero + for i, k := range imp.FcapKeys { + off := binKeysOffset + i*8 + binary.LittleEndian.PutUint64(buf[off:], HashKey(k)) + } + // Zero remaining slots so unused hashes don't accidentally match a real key. + for i := len(imp.FcapKeys); i < MaxKeysPerImpression; i++ { + off := binKeysOffset + i*8 + binary.LittleEndian.PutUint64(buf[off:], 0) + } +} + +func newBinaryBuf(numEntries int) []byte { + buf := make([]byte, binHeaderSize, binHeaderSize+numEntries*binEntrySize) + binary.LittleEndian.PutUint16(buf[0:], binVersion) + binary.LittleEndian.PutUint16(buf[2:], binEntrySize) + return buf +} + +func entryCount(b []byte) int { + if len(b) < binHeaderSize { + return 0 + } + return (len(b) - binHeaderSize) / binEntrySize +} + +func entryAt(b []byte, i int) []byte { + off := binHeaderSize + i*binEntrySize + return b[off : off+binEntrySize] +} + +// Seed bulk-writes a user's full log in one SET. Used by the bench harness +// to populate steady-state without paying the per-impression RMW cost. +func (s *BinaryStore) Seed(ctx context.Context, userID string, imps []Impression) error { + buf := newBinaryBuf(len(imps)) + buf = buf[:binHeaderSize+len(imps)*binEntrySize] + for i, imp := range imps { + encodeEntry(entryAt(buf, i), imp) + } + return s.rdb.Set(ctx, binaryKey(userID), string(buf), 0).Err() +} + +// Write performs read-modify-write for a single impression: fetches the +// existing log, appends the new entry, writes it back. This is the pattern +// `targeting/exposure_binary.go` uses today. +func (s *BinaryStore) Write(ctx context.Context, userID string, imp Impression) error { + key := binaryKey(userID) + cur, err := s.rdb.Get(ctx, key).Bytes() + if err != nil && !errors.Is(err, redis.Nil) { + return err + } + var n int + if len(cur) >= binHeaderSize { + n = entryCount(cur) + } + out := newBinaryBuf(n + 1) + if n > 0 { + out = append(out, cur[binHeaderSize:]...) + } + out = append(out, make([]byte, binEntrySize)...) + encodeEntry(out[binHeaderSize+n*binEntrySize:], imp) + return s.rdb.Set(ctx, key, string(out), 0).Err() +} + +// ReadAndCheck fetches the user's log and answers a single eligibility +// check: does any rule's count of distinct impressions matching fcapKeyHash +// in its window exceed MaxCount? Also returns the latest matching timestamp +// for intent score. +func (s *BinaryStore) ReadAndCheck(ctx context.Context, userID string, fcapKeyHash uint64, rules []FrequencyRule, now int64) (capped bool, latestTS int64, err error) { + b, err := s.rdb.Get(ctx, binaryKey(userID)).Bytes() + if err != nil { + if errors.Is(err, redis.Nil) { + return false, 0, nil + } + return false, 0, err + } + n := entryCount(b) + for _, rule := range rules { + cutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for i := 0; i < n; i++ { + e := entryAt(b, i) + ts := int64(binary.LittleEndian.Uint64(e[binTSOffset:])) //nolint:gosec + if ts < cutoff { + continue + } + kc := int(e[binCountOffset]) + match := false + for j := 0; j < kc; j++ { + if binary.LittleEndian.Uint64(e[binKeysOffset+j*8:]) == fcapKeyHash { + match = true + break + } + } + if !match { + continue + } + impHash := binary.LittleEndian.Uint64(e[binImpOffset:]) + if _, dup := seen[impHash]; dup { + continue + } + seen[impHash] = struct{}{} + count++ + if ts > latestTS { + latestTS = ts + } + } + if count >= rule.MaxCount { + capped = true + } + } + return capped, latestTS, nil +} + +// ReadBatchCheck fetches once, scans once, and answers eligibility for many +// fcap_keys at once by bucketing entries per key. Mirrors the structure of +// `targeting/exposure_aggregate.go` (the PR #103 preagg). +func (s *BinaryStore) ReadBatchCheck(ctx context.Context, userID string, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) (cappedByKey map[uint64]bool, err error) { + b, err := s.rdb.Get(ctx, binaryKey(userID)).Bytes() + if err != nil && !errors.Is(err, redis.Nil) { + return nil, err + } + cappedByKey = make(map[uint64]bool, len(fcapKeyHashes)) + if len(b) < binHeaderSize { + return cappedByKey, nil + } + n := entryCount(b) + + type aggEntry struct { + impHash uint64 + ts int64 + } + // Pre-bucket entries by fcap_key hash. Same idea as exposure_aggregate.go. + wantedSet := make(map[uint64]struct{}, len(fcapKeyHashes)) + for _, k := range fcapKeyHashes { + wantedSet[k] = struct{}{} + } + byKey := make(map[uint64][]aggEntry, len(fcapKeyHashes)) + for i := 0; i < n; i++ { + e := entryAt(b, i) + ts := int64(binary.LittleEndian.Uint64(e[binTSOffset:])) //nolint:gosec + impHash := binary.LittleEndian.Uint64(e[binImpOffset:]) + kc := int(e[binCountOffset]) + for j := 0; j < kc; j++ { + kh := binary.LittleEndian.Uint64(e[binKeysOffset+j*8:]) + if _, want := wantedSet[kh]; !want { + continue + } + byKey[kh] = append(byKey[kh], aggEntry{impHash, ts}) + } + } + + for _, kh := range fcapKeyHashes { + bucket := byKey[kh] + capped := false + for _, rule := range rules { + cutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for _, e := range bucket { + if e.ts < cutoff { + continue + } + if _, dup := seen[e.impHash]; dup { + continue + } + seen[e.impHash] = struct{}{} + count++ + } + if count >= rule.MaxCount { + capped = true + break + } + } + cappedByKey[kh] = capped + } + return cappedByKey, nil +} + +// Cleanup drops entries with timestamps older than (now - window). Performs +// a full read-modify-write of the slab. +func (s *BinaryStore) Cleanup(ctx context.Context, userID string, now int64, window time.Duration) error { + key := binaryKey(userID) + b, err := s.rdb.Get(ctx, key).Bytes() + if err != nil { + if errors.Is(err, redis.Nil) { + return nil + } + return err + } + cutoff := now - int64(window.Seconds()) + n := entryCount(b) + // Find the first kept entry. Entries are append-ordered so timestamps + // are roughly monotonic; we still scan to be safe. + keepFrom := n + for i := 0; i < n; i++ { + ts := int64(binary.LittleEndian.Uint64(entryAt(b, i)[binTSOffset:])) //nolint:gosec + if ts >= cutoff { + keepFrom = i + break + } + } + if keepFrom == 0 { + return nil + } + kept := n - keepFrom + out := newBinaryBuf(kept) + out = out[:binHeaderSize+kept*binEntrySize] + copy(out[binHeaderSize:], b[binHeaderSize+keepFrom*binEntrySize:]) + return s.rdb.Set(ctx, key, string(out), 0).Err() +} + +// MemoryUsage returns the byte size of the user's log in valkey. +func (s *BinaryStore) MemoryUsage(ctx context.Context, userID string) (int64, error) { + return s.rdb.MemoryUsage(ctx, binaryKey(userID)).Result() +} + +// Reset deletes all bench data for this user. +func (s *BinaryStore) Reset(ctx context.Context, userID string) error { + return s.rdb.Del(ctx, binaryKey(userID)).Err() +} diff --git a/targeting/expbench/bucket_count.go b/targeting/expbench/bucket_count.go new file mode 100644 index 00000000..355f80c8 --- /dev/null +++ b/targeting/expbench/bucket_count.go @@ -0,0 +1,262 @@ +package expbench + +import ( + "context" + "errors" + "fmt" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +// Bucket-count variant: HASH per (user, UTC-day). Fields are fcap_key +// hashes (hex-encoded for HASH-friendly strings); values are integer +// counters incremented per impression. Models the "N per day" rule shape +// where N > 1. +// +// Key: user:exp:bc:{uid}:{utc_day} +// Field: hex(fcap_key_hash) +// Value: count +// TTL: 25 hours on the whole HASH +// +// Eligibility under MaxCount=N: +// - HGET for single-key read: capped iff count >= N +// - HMGET for batch read: one RT, returns counts for all candidates +// +// Same period-bucketing pattern as bucket-day but supports counters +// instead of presence-only. + +type BucketCountStore struct { + rdb redis.Cmdable +} + +func NewBucketCountStore(rdb redis.Cmdable) *BucketCountStore { return &BucketCountStore{rdb: rdb} } + +func (s *BucketCountStore) Name() string { return "bucket-count" } + +func bucketCountKey(userID string, dayEpoch int64) string { + return fmt.Sprintf("user:exp:bc:%s:%d", userID, dayEpoch) +} + +func fieldName(h uint64) string { + return strconv.FormatUint(h, 16) +} + +// Seed bulk-writes the user's history. Each impression's K fcap_keys are +// HINCRBY'd in the HASH keyed by that impression's UTC day, plus a single +// EXPIRE per affected day. +func (s *BucketCountStore) Seed(ctx context.Context, userID string, imps []Impression) error { + if len(imps) == 0 { + return nil + } + pipe := s.rdb.Pipeline() + flushEvery := 0 + expiredKeys := make(map[string]struct{}) + for _, imp := range imps { + day := utcDay(imp.Timestamp) + key := bucketCountKey(userID, day) + for _, k := range imp.FcapKeys { + pipe.HIncrBy(ctx, key, fieldName(HashKey(k)), 1) + } + if _, set := expiredKeys[key]; !set { + pipe.Expire(ctx, key, 25*time.Hour) + expiredKeys[key] = struct{}{} + } + flushEvery++ + if flushEvery%1000 == 0 { + if _, err := pipe.Exec(ctx); err != nil { + return err + } + pipe = s.rdb.Pipeline() + } + } + _, err := pipe.Exec(ctx) + return err +} + +// Write writes a single impression. K HINCRBY + 1 EXPIRE pipelined. +func (s *BucketCountStore) Write(ctx context.Context, userID string, imp Impression) error { + day := utcDay(imp.Timestamp) + key := bucketCountKey(userID, day) + if len(imp.FcapKeys) == 0 { + return nil + } + pipe := s.rdb.Pipeline() + for _, k := range imp.FcapKeys { + pipe.HIncrBy(ctx, key, fieldName(HashKey(k)), 1) + } + pipe.Expire(ctx, key, 25*time.Hour) + _, err := pipe.Exec(ctx) + return err +} + +// ReadAndCheck answers "has the count for this fcap_key in today's bucket +// reached MaxCount?" +func (s *BucketCountStore) ReadAndCheck(ctx context.Context, userID string, fcapKeyHash uint64, rules []FrequencyRule, now int64) (capped bool, latestTS int64, err error) { + day := utcDay(now) + key := bucketCountKey(userID, day) + val, err := s.rdb.HGet(ctx, key, fieldName(fcapKeyHash)).Int64() + if err != nil { + if errors.Is(err, redis.Nil) { + return false, 0, nil + } + return false, 0, err + } + for _, rule := range rules { + if val >= int64(rule.MaxCount) { + return true, 0, nil + } + } + return false, 0, nil +} + +// ReadBatchCheck answers eligibility for many fcap_keys. Single HGETALL +// returns all of the user's seen-counts for today; client-side intersects +// with the candidate keys and compares to MaxCount. This matches the +// shape of bucket-day's SMEMBERS+intersect pattern: one RT, response +// size scales with what's actually in the bucket, not the request batch. +func (s *BucketCountStore) ReadBatchCheck(ctx context.Context, userID string, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) (cappedByKey map[uint64]bool, err error) { + day := utcDay(now) + key := bucketCountKey(userID, day) + all, err := s.rdb.HGetAll(ctx, key).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return nil, err + } + counts := make(map[uint64]int64, len(all)) + for f, v := range all { + kh, err := strconv.ParseUint(f, 16, 64) + if err != nil { + continue + } + c, err := strconv.ParseInt(v, 10, 64) + if err != nil { + continue + } + counts[kh] = c + } + maxCount := 0 + for _, rule := range rules { + if rule.MaxCount > maxCount { + maxCount = rule.MaxCount + } + } + cappedByKey = make(map[uint64]bool, len(fcapKeyHashes)) + for _, kh := range fcapKeyHashes { + cappedByKey[kh] = counts[kh] >= int64(maxCount) + } + return cappedByKey, nil +} + +// Cleanup is a no-op — TTL handles it. SCAN walk for symmetry with other +// variants' bench timing. +func (s *BucketCountStore) Cleanup(ctx context.Context, userID string, now int64, window time.Duration) error { + cursor := uint64(0) + pattern := "user:exp:bc:" + userID + ":*" + for { + _, next, err := s.rdb.Scan(ctx, cursor, pattern, 100).Result() + if err != nil { + return err + } + cursor = next + if cursor == 0 { + break + } + } + return nil +} + +// MemoryUsage sums valkey-reported memory for all of the user's per-day +// HASHes. +func (s *BucketCountStore) MemoryUsage(ctx context.Context, userID string) (int64, error) { + pattern := "user:exp:bc:" + userID + ":*" + var total int64 + cursor := uint64(0) + for { + keys, next, err := s.rdb.Scan(ctx, cursor, pattern, 100).Result() + if err != nil { + return total, err + } + for _, k := range keys { + if u, err := s.rdb.MemoryUsage(ctx, k).Result(); err == nil { + total += u + } + } + cursor = next + if cursor == 0 { + break + } + } + return total, nil +} + +// Reset deletes all of the user's per-day HASHes. +func (s *BucketCountStore) Reset(ctx context.Context, userID string) error { + pattern := "user:exp:bc:" + userID + ":*" + cursor := uint64(0) + for { + keys, next, err := s.rdb.Scan(ctx, cursor, pattern, 100).Result() + if err != nil { + return err + } + if len(keys) > 0 { + if err := s.rdb.Del(ctx, keys...).Err(); err != nil { + return err + } + } + cursor = next + if cursor == 0 { + break + } + } + return nil +} + +// --- Split bench support --- + +// Fetch for the bucket-count variant: HGETALL on today's HASH. Re-encodes +// the result as redis.Z entries so the existing FetchResult plumbing works. +// Score = count, Member = hex(fcap_key_hash). +func (s *BucketCountStore) Fetch(ctx context.Context, userID string, window time.Duration, now int64) (FetchResult, error) { + day := utcDay(now) + key := bucketCountKey(userID, day) + res, err := s.rdb.HGetAll(ctx, key).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return FetchResult{}, err + } + zs := make([]redis.Z, 0, len(res)) + for f, v := range res { + count, _ := strconv.ParseFloat(v, 64) + zs = append(zs, redis.Z{Score: count, Member: f}) + } + return FetchResult{ZMembers: zs}, nil +} + +// Process for the bucket-count variant: count map → check against MaxCount. +func (s *BucketCountStore) Process(raw FetchResult, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) map[uint64]bool { + counts := make(map[uint64]int64, len(raw.ZMembers)) + for _, z := range raw.ZMembers { + field, ok := z.Member.(string) + if !ok { + continue + } + kh, err := strconv.ParseUint(field, 16, 64) + if err != nil { + continue + } + counts[kh] = int64(z.Score) + } + maxCount := 0 + for _, rule := range rules { + if rule.MaxCount > maxCount { + maxCount = rule.MaxCount + } + } + out := make(map[uint64]bool, len(fcapKeyHashes)) + for _, kh := range fcapKeyHashes { + out[kh] = counts[kh] >= int64(maxCount) + } + return out +} + +var _ SplitVariant = (*BucketCountStore)(nil) diff --git a/targeting/expbench/bucket_day.go b/targeting/expbench/bucket_day.go new file mode 100644 index 00000000..77e1fe12 --- /dev/null +++ b/targeting/expbench/bucket_day.go @@ -0,0 +1,264 @@ +package expbench + +import ( + "context" + "encoding/binary" + "errors" + "fmt" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +// Bucket-day variant: SET per (user, UTC-day). Members are fcap_key +// hashes the user has seen on that day. Models the "1 per day" rule shape +// from the AppNexus-style bucket model — no exposure log, no scan, no +// preagg index. +// +// Key: user:exp:bd:{uid}:{utc_day} +// Member: fcap_key_hash (8 bytes) +// TTL: 25 hours (set on every SADD; idempotent EXPIRE is cheap) +// +// Eligibility under MaxCount=1 (singleton-per-day): +// - SISMEMBER for single-key read +// - SMEMBERS + client-side intersect for batch read (one RT regardless +// of batch size) +// +// MaxCount>1 ("5 per day") would use the same shape with HASH fields and +// HINCRBY; not implemented here. Keeping the comparison focused on the +// dominant singleton case. + +type BucketDayStore struct { + rdb redis.Cmdable +} + +func NewBucketDayStore(rdb redis.Cmdable) *BucketDayStore { return &BucketDayStore{rdb: rdb} } + +func (s *BucketDayStore) Name() string { return "bucket-day" } + +func bucketDayKey(userID string, dayEpoch int64) string { + return fmt.Sprintf("user:exp:bd:%s:%d", userID, dayEpoch) +} + +// utcDay returns the UTC day index (days since epoch) for the given unix +// timestamp. Daily buckets are simple integer division. +func utcDay(unixSec int64) int64 { + return unixSec / 86400 +} + +// memberBytes encodes an 8-byte hash as a SET member. +func memberBytes(h uint64) string { + buf := make([]byte, 8) + binary.LittleEndian.PutUint64(buf, h) + return string(buf) +} + +func decodeMemberBytes(m string) uint64 { + if len(m) < 8 { + return 0 + } + return binary.LittleEndian.Uint64([]byte(m)) +} + +// Seed bulk-writes the user's history. Each impression's K fcap_keys are +// SADDed to the SET keyed by that impression's UTC day. Older days are +// included even though TTL would have expired them in production — for +// the bench we want all days populated to measure realistic membership +// counts. (This means the bench measures "fully-populated 30 days of +// buckets," matching how the ZSET variants are seeded.) +func (s *BucketDayStore) Seed(ctx context.Context, userID string, imps []Impression) error { + if len(imps) == 0 { + return nil + } + pipe := s.rdb.Pipeline() + flushEvery := 0 + for _, imp := range imps { + day := utcDay(imp.Timestamp) + key := bucketDayKey(userID, day) + members := make([]any, 0, len(imp.FcapKeys)) + for _, k := range imp.FcapKeys { + members = append(members, memberBytes(HashKey(k))) + } + pipe.SAdd(ctx, key, members...) + pipe.Expire(ctx, key, 25*time.Hour) + flushEvery++ + if flushEvery%1000 == 0 { + if _, err := pipe.Exec(ctx); err != nil { + return err + } + pipe = s.rdb.Pipeline() + } + } + _, err := pipe.Exec(ctx) + return err +} + +// Write writes a single impression to today's bucket. K hashes SADDed + +// 25h EXPIRE, all in one RT. +func (s *BucketDayStore) Write(ctx context.Context, userID string, imp Impression) error { + day := utcDay(imp.Timestamp) + key := bucketDayKey(userID, day) + members := make([]any, 0, len(imp.FcapKeys)) + for _, k := range imp.FcapKeys { + members = append(members, memberBytes(HashKey(k))) + } + if len(members) == 0 { + return nil + } + pipe := s.rdb.Pipeline() + pipe.SAdd(ctx, key, members...) + pipe.Expire(ctx, key, 25*time.Hour) + _, err := pipe.Exec(ctx) + return err +} + +// ReadAndCheck answers "is this fcap_key in today's bucket?" Treats the +// rule as a singleton-per-day rule: capped iff member is present. +// MaxCount > 1 is not modeled here. +func (s *BucketDayStore) ReadAndCheck(ctx context.Context, userID string, fcapKeyHash uint64, rules []FrequencyRule, now int64) (capped bool, latestTS int64, err error) { + day := utcDay(now) + key := bucketDayKey(userID, day) + present, err := s.rdb.SIsMember(ctx, key, memberBytes(fcapKeyHash)).Result() + if err != nil { + if errors.Is(err, redis.Nil) { + return false, 0, nil + } + return false, 0, err + } + // latestTS is irrelevant in bucket model — bucket only carries presence, + // not timestamps. Return 0; intent score in production would derive from + // a separate signal if needed. + if present { + return true, 0, nil + } + return false, 0, nil +} + +// ReadBatchCheck answers eligibility for many fcap_keys with a single +// SMEMBERS + client-side intersect. Single RT regardless of batch size. +func (s *BucketDayStore) ReadBatchCheck(ctx context.Context, userID string, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) (cappedByKey map[uint64]bool, err error) { + day := utcDay(now) + key := bucketDayKey(userID, day) + res, err := s.rdb.SMembers(ctx, key).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return nil, err + } + seen := make(map[uint64]struct{}, len(res)) + for _, m := range res { + seen[decodeMemberBytes(m)] = struct{}{} + } + cappedByKey = make(map[uint64]bool, len(fcapKeyHashes)) + for _, kh := range fcapKeyHashes { + _, present := seen[kh] + cappedByKey[kh] = present + } + return cappedByKey, nil +} + +// Cleanup is a no-op in bucket-day — TTL handles it server-side. Kept on +// the interface for symmetry; the call costs essentially nothing. +func (s *BucketDayStore) Cleanup(ctx context.Context, userID string, now int64, window time.Duration) error { + // SCAN + DEL would forcibly remove old buckets; in practice TTL handles + // it. Run a no-op SCAN so the bench has something to measure for the + // "cleanup" column. + cursor := uint64(0) + pattern := "user:exp:bd:" + userID + ":*" + for { + keys, next, err := s.rdb.Scan(ctx, cursor, pattern, 100).Result() + if err != nil { + return err + } + _ = keys // we don't actually delete; TTL handles it + cursor = next + if cursor == 0 { + break + } + } + return nil +} + +// MemoryUsage sums valkey-reported memory for all of the user's per-day +// SETs (discovered via SCAN). +func (s *BucketDayStore) MemoryUsage(ctx context.Context, userID string) (int64, error) { + pattern := "user:exp:bd:" + userID + ":*" + var total int64 + cursor := uint64(0) + for { + keys, next, err := s.rdb.Scan(ctx, cursor, pattern, 100).Result() + if err != nil { + return total, err + } + for _, k := range keys { + if u, err := s.rdb.MemoryUsage(ctx, k).Result(); err == nil { + total += u + } + } + cursor = next + if cursor == 0 { + break + } + } + return total, nil +} + +// Reset deletes all of the user's per-day buckets. +func (s *BucketDayStore) Reset(ctx context.Context, userID string) error { + pattern := "user:exp:bd:" + userID + ":*" + cursor := uint64(0) + for { + keys, next, err := s.rdb.Scan(ctx, cursor, pattern, 100).Result() + if err != nil { + return err + } + if len(keys) > 0 { + if err := s.rdb.Del(ctx, keys...).Err(); err != nil { + return err + } + } + cursor = next + if cursor == 0 { + break + } + } + return nil +} + +// --- Split bench support --- + +// Fetch for the bucket variant: SMEMBERS on today's bucket. Returns the +// members in FetchResult.ZMembers (re-using the field; score is unused). +func (s *BucketDayStore) Fetch(ctx context.Context, userID string, window time.Duration, now int64) (FetchResult, error) { + day := utcDay(now) + key := bucketDayKey(userID, day) + res, err := s.rdb.SMembers(ctx, key).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return FetchResult{}, err + } + zs := make([]redis.Z, len(res)) + for i, m := range res { + zs[i] = redis.Z{Member: m} + } + return FetchResult{ZMembers: zs}, nil +} + +// Process for the bucket variant: build a hash set from members, +// intersect with candidates. +func (s *BucketDayStore) Process(raw FetchResult, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) map[uint64]bool { + seen := make(map[uint64]struct{}, len(raw.ZMembers)) + for _, z := range raw.ZMembers { + seen[decodeMemberBytes(z.Member.(string))] = struct{}{} + } + out := make(map[uint64]bool, len(fcapKeyHashes)) + for _, kh := range fcapKeyHashes { + _, present := seen[kh] + out[kh] = present + } + return out +} + +// Compile-time check. +var _ SplitVariant = (*BucketDayStore)(nil) + +// avoid unused import in linters +var _ = strconv.Itoa diff --git a/targeting/expbench/go.mod b/targeting/expbench/go.mod new file mode 100644 index 00000000..931ceea2 --- /dev/null +++ b/targeting/expbench/go.mod @@ -0,0 +1,10 @@ +module github.com/adcontextprotocol/adcp-go/targeting/expbench + +go 1.25 + +require github.com/redis/go-redis/v9 v9.7.0 + +require ( + github.com/cespare/xxhash/v2 v2.2.0 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect +) diff --git a/targeting/expbench/go.sum b/targeting/expbench/go.sum new file mode 100644 index 00000000..f11d99f0 --- /dev/null +++ b/targeting/expbench/go.sum @@ -0,0 +1,10 @@ +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= +github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= +github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E= +github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw= diff --git a/targeting/expbench/split.go b/targeting/expbench/split.go new file mode 100644 index 00000000..e694123f --- /dev/null +++ b/targeting/expbench/split.go @@ -0,0 +1,239 @@ +package expbench + +import ( + "context" + "encoding/binary" + "errors" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +// FetchResult carries the raw, pre-Go-processing data from a single fetch. +// It's variant-specific behind an opaque type so the bench can time the +// fetch (valkey-side cost + network + Go-side deserialization of the wire +// protocol) separately from the in-process eligibility scan. +type FetchResult struct { + BinaryBlob []byte // BinaryStore + ZMembers []redis.Z // ZSet variants +} + +// Fetcher pulls raw data for a window from valkey. Implemented by each +// variant alongside its existing Read* methods so the bench can split +// "wait on valkey" from "scan in Go". +type Fetcher interface { + Fetch(ctx context.Context, userID string, window time.Duration, now int64) (FetchResult, error) +} + +// Processor runs the Go-side eligibility scan against pre-fetched data. +// No I/O — pure CPU. +type Processor interface { + Process(raw FetchResult, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) (cappedByKey map[uint64]bool) +} + +// --- BinaryStore --- + +func (s *BinaryStore) Fetch(ctx context.Context, userID string, window time.Duration, now int64) (FetchResult, error) { + b, err := s.rdb.Get(ctx, binaryKey(userID)).Bytes() + if err != nil && !errors.Is(err, redis.Nil) { + return FetchResult{}, err + } + return FetchResult{BinaryBlob: b}, nil +} + +func (s *BinaryStore) Process(raw FetchResult, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) map[uint64]bool { + b := raw.BinaryBlob + cappedByKey := make(map[uint64]bool, len(fcapKeyHashes)) + if len(b) < binHeaderSize { + return cappedByKey + } + n := entryCount(b) + type aggEntry struct { + impHash uint64 + ts int64 + } + wantedSet := make(map[uint64]struct{}, len(fcapKeyHashes)) + for _, k := range fcapKeyHashes { + wantedSet[k] = struct{}{} + } + byKey := make(map[uint64][]aggEntry, len(fcapKeyHashes)) + for i := 0; i < n; i++ { + e := entryAt(b, i) + ts := int64(binary.LittleEndian.Uint64(e[binTSOffset:])) //nolint:gosec + impHash := binary.LittleEndian.Uint64(e[binImpOffset:]) + kc := int(e[binCountOffset]) + for j := 0; j < kc; j++ { + kh := binary.LittleEndian.Uint64(e[binKeysOffset+j*8:]) + if _, want := wantedSet[kh]; !want { + continue + } + byKey[kh] = append(byKey[kh], aggEntry{impHash, ts}) + } + } + for _, kh := range fcapKeyHashes { + bucket := byKey[kh] + capped := false + for _, rule := range rules { + cutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for _, e := range bucket { + if e.ts < cutoff { + continue + } + if _, dup := seen[e.impHash]; dup { + continue + } + seen[e.impHash] = struct{}{} + count++ + } + if count >= rule.MaxCount { + capped = true + break + } + } + cappedByKey[kh] = capped + } + return cappedByKey +} + +// --- ZSetArrayStore --- + +func (s *ZSetArrayStore) Fetch(ctx context.Context, userID string, window time.Duration, now int64) (FetchResult, error) { + cutoff := now - int64(window.Seconds()) + res, err := s.rdb.ZRangeByScoreWithScores(ctx, zsetArrayKey(userID), &redis.ZRangeBy{ + Min: strconv.FormatInt(cutoff, 10), + Max: "+inf", + }).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return FetchResult{}, err + } + return FetchResult{ZMembers: res}, nil +} + +func (s *ZSetArrayStore) Process(raw FetchResult, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) map[uint64]bool { + wantedSet := make(map[uint64]struct{}, len(fcapKeyHashes)) + for _, k := range fcapKeyHashes { + wantedSet[k] = struct{}{} + } + type entry struct { + impHash uint64 + ts int64 + } + byKey := make(map[uint64][]entry, len(fcapKeyHashes)) + for _, z := range raw.ZMembers { + impHash, keyHashes := decodeArrayMember(z.Member.(string)) + ts := int64(z.Score) + for _, kh := range keyHashes { + if _, want := wantedSet[kh]; !want { + continue + } + byKey[kh] = append(byKey[kh], entry{impHash, ts}) + } + } + cappedByKey := make(map[uint64]bool, len(fcapKeyHashes)) + for _, kh := range fcapKeyHashes { + bucket := byKey[kh] + capped := false + for _, rule := range rules { + cutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for _, e := range bucket { + if e.ts < cutoff { + continue + } + if _, dup := seen[e.impHash]; dup { + continue + } + seen[e.impHash] = struct{}{} + count++ + } + if count >= rule.MaxCount { + capped = true + break + } + } + cappedByKey[kh] = capped + } + return cappedByKey +} + +// --- ZSetPerKeyStore --- + +func (s *ZSetPerKeyStore) Fetch(ctx context.Context, userID string, window time.Duration, now int64) (FetchResult, error) { + cutoff := now - int64(window.Seconds()) + res, err := s.rdb.ZRangeByScoreWithScores(ctx, zsetPerKeyKey(userID), &redis.ZRangeBy{ + Min: strconv.FormatInt(cutoff, 10), + Max: "+inf", + }).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return FetchResult{}, err + } + return FetchResult{ZMembers: res}, nil +} + +func (s *ZSetPerKeyStore) Process(raw FetchResult, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) map[uint64]bool { + wantedSet := make(map[uint64]struct{}, len(fcapKeyHashes)) + for _, k := range fcapKeyHashes { + wantedSet[k] = struct{}{} + } + type entry struct { + impHash uint64 + ts int64 + } + byKey := make(map[uint64][]entry, len(fcapKeyHashes)) + for _, z := range raw.ZMembers { + impHash, keyHash := decodePerKeyMember(z.Member.(string)) + if _, want := wantedSet[keyHash]; !want { + continue + } + byKey[keyHash] = append(byKey[keyHash], entry{impHash, int64(z.Score)}) + } + cappedByKey := make(map[uint64]bool, len(fcapKeyHashes)) + for _, kh := range fcapKeyHashes { + bucket := byKey[kh] + capped := false + for _, rule := range rules { + cutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for _, e := range bucket { + if e.ts < cutoff { + continue + } + if _, dup := seen[e.impHash]; dup { + continue + } + seen[e.impHash] = struct{}{} + count++ + } + if count >= rule.MaxCount { + capped = true + break + } + } + cappedByKey[kh] = capped + } + return cappedByKey +} + +// SplitVariant combines Fetcher + Processor + the Variant identity needed +// for bench iteration. +type SplitVariant interface { + Variant + Fetcher + Processor +} + +// SizeOfFetch is a rough indicator of how many bytes/members came back — +// useful for understanding why one variant's fetch costs more than another. +func SizeOfFetch(r FetchResult) (bytes int, members int) { + bytes = len(r.BinaryBlob) + members = len(r.ZMembers) + for _, z := range r.ZMembers { + bytes += len(z.Member.(string)) + } + return bytes, members +} diff --git a/targeting/expbench/types.go b/targeting/expbench/types.go new file mode 100644 index 00000000..4981ae04 --- /dev/null +++ b/targeting/expbench/types.go @@ -0,0 +1,48 @@ +// Package expbench compares three exposure-log storage shapes against valkey: +// a generalized fixed-stride binary log, a ZSET with fcap_keys array per +// member, and a ZSET with one member per fcap_key. The goal is empirical +// answers to "which shape is right for the per-user exposure log under the +// fcap_keys[] data model?" +package expbench + +import ( + "hash/fnv" + "time" +) + +// Impression is one ad delivery to one user, carrying the fcap_keys the +// buyer wants to count this impression against. +type Impression struct { + ImpressionID string + Timestamp int64 + FcapKeys []string // e.g. {"campaign:42", "advertiser:13", "creative:8"} +} + +// FrequencyRule limits exposures matching a single fcap_key in a window. +type FrequencyRule struct { + MaxCount int + Window time.Duration +} + +// MaxKeysPerImpression caps how many fcap_keys a single impression can +// carry in the binary-log variant. ZSET variants are unbounded. +const MaxKeysPerImpression = 8 + +// HashKey returns the 8-byte hash of a label string. Same FNV-1a as +// targeting.hashString — one hash function across all variants so the +// comparison isn't muddied by hash-function differences. +func HashKey(s string) uint64 { + h := fnv.New64a() + _, _ = h.Write([]byte(s)) + return h.Sum64() +} + +// HashKeys hashes a slice of labels in order. Bench helper; production +// would hash at write time. +func HashKeys(keys []string) []uint64 { + out := make([]uint64, len(keys)) + for i, k := range keys { + out[i] = HashKey(k) + } + return out +} diff --git a/targeting/expbench/variant.go b/targeting/expbench/variant.go new file mode 100644 index 00000000..a92d8329 --- /dev/null +++ b/targeting/expbench/variant.go @@ -0,0 +1,36 @@ +package expbench + +import ( + "context" + "time" +) + +// Variant is the common interface implemented by each storage shape so the +// bench harness can iterate over them uniformly. +type Variant interface { + Name() string + + // Seed bulk-writes a user's full log. Used to populate steady-state + // before running per-operation latency loops. + Seed(ctx context.Context, userID string, imps []Impression) error + + // Write writes one impression. Returns latency by virtue of being timed + // by the caller. + Write(ctx context.Context, userID string, imp Impression) error + + // ReadAndCheck answers a single eligibility question. + ReadAndCheck(ctx context.Context, userID string, fcapKeyHash uint64, rules []FrequencyRule, now int64) (capped bool, latestTS int64, err error) + + // ReadBatchCheck answers eligibility for many fcap_keys against the + // same user log in one fetch. + ReadBatchCheck(ctx context.Context, userID string, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) (cappedByKey map[uint64]bool, err error) + + // Cleanup drops entries older than (now - window). + Cleanup(ctx context.Context, userID string, now int64, window time.Duration) error + + // MemoryUsage returns the byte size of the user's log in valkey. + MemoryUsage(ctx context.Context, userID string) (int64, error) + + // Reset deletes all bench data for this user. + Reset(ctx context.Context, userID string) error +} diff --git a/targeting/expbench/zset_array.go b/targeting/expbench/zset_array.go new file mode 100644 index 00000000..0426a1d7 --- /dev/null +++ b/targeting/expbench/zset_array.go @@ -0,0 +1,231 @@ +package expbench + +import ( + "context" + "encoding/binary" + "errors" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +// ZSet-array variant: one ZSET per user. Each ZADD writes one member that +// encodes a single impression with all of its fcap_key hashes inline. +// +// Member layout: impHash(8) + keyCount(1) + fcapKeyHash[0..keyCount-1](8 each) +// Score: timestamp (unix seconds, stored as float64). +// +// Read uses ZRANGEBYSCORE for server-side window filtering. + +type ZSetArrayStore struct { + rdb redis.Cmdable +} + +func NewZSetArrayStore(rdb redis.Cmdable) *ZSetArrayStore { return &ZSetArrayStore{rdb: rdb} } + +func (s *ZSetArrayStore) Name() string { return "zset-array" } + +func zsetArrayKey(userID string) string { return "user:exposures:zsa:" + userID } + +func encodeArrayMember(imp Impression) string { + keys := imp.FcapKeys + if len(keys) > MaxKeysPerImpression { + keys = keys[:MaxKeysPerImpression] + } + buf := make([]byte, 9+len(keys)*8) + binary.LittleEndian.PutUint64(buf[0:], HashKey(imp.ImpressionID)) + buf[8] = byte(len(keys)) + for i, k := range keys { + binary.LittleEndian.PutUint64(buf[9+i*8:], HashKey(k)) + } + return string(buf) +} + +func decodeArrayMember(m string) (impHash uint64, keyHashes []uint64) { + if len(m) < 9 { + return 0, nil + } + b := []byte(m) + impHash = binary.LittleEndian.Uint64(b[0:]) + kc := int(b[8]) + if 9+kc*8 > len(b) { + return impHash, nil + } + keyHashes = make([]uint64, kc) + for i := 0; i < kc; i++ { + keyHashes[i] = binary.LittleEndian.Uint64(b[9+i*8:]) + } + return impHash, keyHashes +} + +// Seed bulk-writes a user's full log via a single pipelined ZADD batch. +func (s *ZSetArrayStore) Seed(ctx context.Context, userID string, imps []Impression) error { + if len(imps) == 0 { + return nil + } + key := zsetArrayKey(userID) + zs := make([]redis.Z, 0, len(imps)) + for _, imp := range imps { + zs = append(zs, redis.Z{ + Score: float64(imp.Timestamp), + Member: encodeArrayMember(imp), + }) + } + return s.rdb.ZAdd(ctx, key, zs...).Err() +} + +// Write writes a single impression. One ZADD round-trip. +func (s *ZSetArrayStore) Write(ctx context.Context, userID string, imp Impression) error { + return s.rdb.ZAdd(ctx, zsetArrayKey(userID), redis.Z{ + Score: float64(imp.Timestamp), + Member: encodeArrayMember(imp), + }).Err() +} + +// ReadAndCheck answers a single eligibility check. Server-side window +// filter via ZRANGEBYSCORE; client iterates returned members. +func (s *ZSetArrayStore) ReadAndCheck(ctx context.Context, userID string, fcapKeyHash uint64, rules []FrequencyRule, now int64) (capped bool, latestTS int64, err error) { + if len(rules) == 0 { + return false, 0, nil + } + // Use the widest rule window for the fetch; smaller windows filter on the client. + maxWindow := rules[0].Window + for _, r := range rules[1:] { + if r.Window > maxWindow { + maxWindow = r.Window + } + } + cutoff := now - int64(maxWindow.Seconds()) + res, err := s.rdb.ZRangeByScoreWithScores(ctx, zsetArrayKey(userID), &redis.ZRangeBy{ + Min: strconv.FormatInt(cutoff, 10), + Max: "+inf", + }).Result() + if err != nil { + if errors.Is(err, redis.Nil) { + return false, 0, nil + } + return false, 0, err + } + + type entry struct { + impHash uint64 + ts int64 + } + matched := make([]entry, 0, len(res)/4) + for _, z := range res { + impHash, keyHashes := decodeArrayMember(z.Member.(string)) + for _, kh := range keyHashes { + if kh == fcapKeyHash { + ts := int64(z.Score) + matched = append(matched, entry{impHash, ts}) + if ts > latestTS { + latestTS = ts + } + break + } + } + } + for _, rule := range rules { + ruleCutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for _, e := range matched { + if e.ts < ruleCutoff { + continue + } + if _, dup := seen[e.impHash]; dup { + continue + } + seen[e.impHash] = struct{}{} + count++ + } + if count >= rule.MaxCount { + capped = true + } + } + return capped, latestTS, nil +} + +// ReadBatchCheck answers eligibility for many fcap_keys with a single +// ZRANGEBYSCORE fetch. Buckets matching members per key. +func (s *ZSetArrayStore) ReadBatchCheck(ctx context.Context, userID string, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) (cappedByKey map[uint64]bool, err error) { + maxWindow := time.Duration(0) + for _, r := range rules { + if r.Window > maxWindow { + maxWindow = r.Window + } + } + cutoff := now - int64(maxWindow.Seconds()) + res, err := s.rdb.ZRangeByScoreWithScores(ctx, zsetArrayKey(userID), &redis.ZRangeBy{ + Min: strconv.FormatInt(cutoff, 10), + Max: "+inf", + }).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return nil, err + } + + wantedSet := make(map[uint64]struct{}, len(fcapKeyHashes)) + for _, k := range fcapKeyHashes { + wantedSet[k] = struct{}{} + } + type entry struct { + impHash uint64 + ts int64 + } + byKey := make(map[uint64][]entry, len(fcapKeyHashes)) + for _, z := range res { + impHash, keyHashes := decodeArrayMember(z.Member.(string)) + ts := int64(z.Score) + for _, kh := range keyHashes { + if _, want := wantedSet[kh]; !want { + continue + } + byKey[kh] = append(byKey[kh], entry{impHash, ts}) + } + } + + cappedByKey = make(map[uint64]bool, len(fcapKeyHashes)) + for _, kh := range fcapKeyHashes { + bucket := byKey[kh] + capped := false + for _, rule := range rules { + ruleCutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for _, e := range bucket { + if e.ts < ruleCutoff { + continue + } + if _, dup := seen[e.impHash]; dup { + continue + } + seen[e.impHash] = struct{}{} + count++ + } + if count >= rule.MaxCount { + capped = true + break + } + } + cappedByKey[kh] = capped + } + return cappedByKey, nil +} + +// Cleanup drops entries older than (now - window) via a single +// ZREMRANGEBYSCORE. Server-side, no read-modify-write. +func (s *ZSetArrayStore) Cleanup(ctx context.Context, userID string, now int64, window time.Duration) error { + cutoff := now - int64(window.Seconds()) + return s.rdb.ZRemRangeByScore(ctx, zsetArrayKey(userID), "-inf", "("+strconv.FormatInt(cutoff, 10)).Err() +} + +// MemoryUsage returns the byte size of the user's ZSET in valkey. +func (s *ZSetArrayStore) MemoryUsage(ctx context.Context, userID string) (int64, error) { + return s.rdb.MemoryUsage(ctx, zsetArrayKey(userID)).Result() +} + +// Reset deletes all bench data for this user. +func (s *ZSetArrayStore) Reset(ctx context.Context, userID string) error { + return s.rdb.Del(ctx, zsetArrayKey(userID)).Err() +} diff --git a/targeting/expbench/zset_perkey.go b/targeting/expbench/zset_perkey.go new file mode 100644 index 00000000..d43c70aa --- /dev/null +++ b/targeting/expbench/zset_perkey.go @@ -0,0 +1,241 @@ +package expbench + +import ( + "context" + "encoding/binary" + "errors" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +// ZSet-per-key variant: one ZSET per user, but each impression produces +// K members (one per fcap_key). Member is fixed 16 bytes. +// +// Member layout: impHash(8) + fcapKeyHash(8) +// Score: timestamp. +// +// Trade vs. ZSetArrayStore: +// - K× write amplification (K ZADDs per impression, pipelined into one RT) +// - K× more entries per ZSET → K× the memory +// - Read-side: no per-member deserialization beyond the 8/8 byte split; +// impression-level dedup needs a client-side seen-set (one impression +// produces multiple matching members under different fcap_keys). + +type ZSetPerKeyStore struct { + rdb redis.Cmdable +} + +func NewZSetPerKeyStore(rdb redis.Cmdable) *ZSetPerKeyStore { return &ZSetPerKeyStore{rdb: rdb} } + +func (s *ZSetPerKeyStore) Name() string { return "zset-perkey" } + +func zsetPerKeyKey(userID string) string { return "user:exposures:zsk:" + userID } + +func encodePerKeyMember(impHash, keyHash uint64) string { + buf := make([]byte, 16) + binary.LittleEndian.PutUint64(buf[0:], impHash) + binary.LittleEndian.PutUint64(buf[8:], keyHash) + return string(buf) +} + +func decodePerKeyMember(m string) (impHash, keyHash uint64) { + if len(m) < 16 { + return 0, 0 + } + b := []byte(m) + return binary.LittleEndian.Uint64(b[0:]), binary.LittleEndian.Uint64(b[8:]) +} + +// Seed bulk-writes a user's full log. K members emitted per impression. +func (s *ZSetPerKeyStore) Seed(ctx context.Context, userID string, imps []Impression) error { + if len(imps) == 0 { + return nil + } + key := zsetPerKeyKey(userID) + zs := make([]redis.Z, 0, len(imps)*4) + for _, imp := range imps { + impHash := HashKey(imp.ImpressionID) + keys := imp.FcapKeys + if len(keys) > MaxKeysPerImpression { + keys = keys[:MaxKeysPerImpression] + } + for _, k := range keys { + zs = append(zs, redis.Z{ + Score: float64(imp.Timestamp), + Member: encodePerKeyMember(impHash, HashKey(k)), + }) + } + } + // ZAdd in chunks to avoid oversized commands at extreme load. + const chunk = 5000 + for i := 0; i < len(zs); i += chunk { + end := i + chunk + if end > len(zs) { + end = len(zs) + } + if err := s.rdb.ZAdd(ctx, key, zs[i:end]...).Err(); err != nil { + return err + } + } + return nil +} + +// Write writes a single impression. K ZADDs pipelined into one RT. +func (s *ZSetPerKeyStore) Write(ctx context.Context, userID string, imp Impression) error { + keys := imp.FcapKeys + if len(keys) > MaxKeysPerImpression { + keys = keys[:MaxKeysPerImpression] + } + if len(keys) == 0 { + return nil + } + impHash := HashKey(imp.ImpressionID) + zs := make([]redis.Z, 0, len(keys)) + for _, k := range keys { + zs = append(zs, redis.Z{ + Score: float64(imp.Timestamp), + Member: encodePerKeyMember(impHash, HashKey(k)), + }) + } + return s.rdb.ZAdd(ctx, zsetPerKeyKey(userID), zs...).Err() +} + +// ReadAndCheck answers a single eligibility check. +func (s *ZSetPerKeyStore) ReadAndCheck(ctx context.Context, userID string, fcapKeyHash uint64, rules []FrequencyRule, now int64) (capped bool, latestTS int64, err error) { + if len(rules) == 0 { + return false, 0, nil + } + maxWindow := rules[0].Window + for _, r := range rules[1:] { + if r.Window > maxWindow { + maxWindow = r.Window + } + } + cutoff := now - int64(maxWindow.Seconds()) + res, err := s.rdb.ZRangeByScoreWithScores(ctx, zsetPerKeyKey(userID), &redis.ZRangeBy{ + Min: strconv.FormatInt(cutoff, 10), + Max: "+inf", + }).Result() + if err != nil { + if errors.Is(err, redis.Nil) { + return false, 0, nil + } + return false, 0, err + } + + type entry struct { + impHash uint64 + ts int64 + } + matched := make([]entry, 0, len(res)/4) + for _, z := range res { + impHash, keyHash := decodePerKeyMember(z.Member.(string)) + if keyHash != fcapKeyHash { + continue + } + ts := int64(z.Score) + matched = append(matched, entry{impHash, ts}) + if ts > latestTS { + latestTS = ts + } + } + for _, rule := range rules { + ruleCutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for _, e := range matched { + if e.ts < ruleCutoff { + continue + } + if _, dup := seen[e.impHash]; dup { + continue + } + seen[e.impHash] = struct{}{} + count++ + } + if count >= rule.MaxCount { + capped = true + } + } + return capped, latestTS, nil +} + +// ReadBatchCheck answers eligibility for many fcap_keys. +func (s *ZSetPerKeyStore) ReadBatchCheck(ctx context.Context, userID string, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) (cappedByKey map[uint64]bool, err error) { + maxWindow := time.Duration(0) + for _, r := range rules { + if r.Window > maxWindow { + maxWindow = r.Window + } + } + cutoff := now - int64(maxWindow.Seconds()) + res, err := s.rdb.ZRangeByScoreWithScores(ctx, zsetPerKeyKey(userID), &redis.ZRangeBy{ + Min: strconv.FormatInt(cutoff, 10), + Max: "+inf", + }).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return nil, err + } + + wantedSet := make(map[uint64]struct{}, len(fcapKeyHashes)) + for _, k := range fcapKeyHashes { + wantedSet[k] = struct{}{} + } + type entry struct { + impHash uint64 + ts int64 + } + byKey := make(map[uint64][]entry, len(fcapKeyHashes)) + for _, z := range res { + impHash, keyHash := decodePerKeyMember(z.Member.(string)) + if _, want := wantedSet[keyHash]; !want { + continue + } + byKey[keyHash] = append(byKey[keyHash], entry{impHash, int64(z.Score)}) + } + + cappedByKey = make(map[uint64]bool, len(fcapKeyHashes)) + for _, kh := range fcapKeyHashes { + bucket := byKey[kh] + capped := false + for _, rule := range rules { + ruleCutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for _, e := range bucket { + if e.ts < ruleCutoff { + continue + } + if _, dup := seen[e.impHash]; dup { + continue + } + seen[e.impHash] = struct{}{} + count++ + } + if count >= rule.MaxCount { + capped = true + break + } + } + cappedByKey[kh] = capped + } + return cappedByKey, nil +} + +// Cleanup drops entries older than (now - window). +func (s *ZSetPerKeyStore) Cleanup(ctx context.Context, userID string, now int64, window time.Duration) error { + cutoff := now - int64(window.Seconds()) + return s.rdb.ZRemRangeByScore(ctx, zsetPerKeyKey(userID), "-inf", "("+strconv.FormatInt(cutoff, 10)).Err() +} + +// MemoryUsage returns the byte size of the user's ZSET in valkey. +func (s *ZSetPerKeyStore) MemoryUsage(ctx context.Context, userID string) (int64, error) { + return s.rdb.MemoryUsage(ctx, zsetPerKeyKey(userID)).Result() +} + +// Reset deletes all bench data for this user. +func (s *ZSetPerKeyStore) Reset(ctx context.Context, userID string) error { + return s.rdb.Del(ctx, zsetPerKeyKey(userID)).Err() +} diff --git a/targeting/expbench/zset_perkeyed.go b/targeting/expbench/zset_perkeyed.go new file mode 100644 index 00000000..8eea6a85 --- /dev/null +++ b/targeting/expbench/zset_perkeyed.go @@ -0,0 +1,421 @@ +package expbench + +import ( + "context" + "encoding/binary" + "errors" + "fmt" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +// ZSet-per-keyed variant: one ZSET per (user, fcap_key) pair. Each +// impression produces K ZADDs, one per fcap_key, each into a different key. +// Member is just the impression hash (8 bytes) — the fcap_key is the +// valkey key, not part of the member. +// +// Win condition: single-key reads only fetch the data for that one key, +// not the user's whole 30-day log. Loss condition: batch reads need K +// fetches (pipelined); cleanup needs to find all of a user's per-key +// ZSETs. +// +// Cleanup story (resolved): +// - We maintain an index `user:exp:idx:{uid}` (a regular SET) of all +// fcap_key_hashes the user has any data for. ZADD'd to on every +// write; cleaned up by the periodic janitor that runs +// ZREMRANGEBYSCORE on each member key and SREM on emptied ones. +// - Index write cost: 1 SADD per impression per fcap_key (K total), +// pipelined with the ZADDs. SADD is idempotent, so a hot impression +// for a known key is a single SADD command in valkey-side cost. + +type ZSetPerKeyedStore struct { + rdb redis.Cmdable +} + +func NewZSetPerKeyedStore(rdb redis.Cmdable) *ZSetPerKeyedStore { return &ZSetPerKeyedStore{rdb: rdb} } + +func (s *ZSetPerKeyedStore) Name() string { return "zset-perkeyed" } + +func zsetPerKeyedKey(userID string, keyHash uint64) string { + return fmt.Sprintf("user:exposures:zsek:%s:%016x", userID, keyHash) +} +func zsetPerKeyedIndex(userID string) string { return "user:exposures:zsek-idx:" + userID } + +func encodePerKeyedMember(impHash uint64) string { + buf := make([]byte, 8) + binary.LittleEndian.PutUint64(buf, impHash) + return string(buf) +} + +func decodePerKeyedMember(m string) uint64 { + if len(m) < 8 { + return 0 + } + return binary.LittleEndian.Uint64([]byte(m)) +} + +// Seed bulk-writes a user's full log. K members (one per fcap_key on each +// impression) plus K SADDs for the index, all pipelined. +func (s *ZSetPerKeyedStore) Seed(ctx context.Context, userID string, imps []Impression) error { + if len(imps) == 0 { + return nil + } + idx := zsetPerKeyedIndex(userID) + pipe := s.rdb.Pipeline() + indexAdded := make(map[uint64]struct{}) + + for _, imp := range imps { + impHash := HashKey(imp.ImpressionID) + keys := imp.FcapKeys + if len(keys) > MaxKeysPerImpression { + keys = keys[:MaxKeysPerImpression] + } + for _, k := range keys { + kh := HashKey(k) + pipe.ZAdd(ctx, zsetPerKeyedKey(userID, kh), redis.Z{ + Score: float64(imp.Timestamp), + Member: encodePerKeyedMember(impHash), + }) + if _, seen := indexAdded[kh]; !seen { + pipe.SAdd(ctx, idx, strconv.FormatUint(kh, 16)) + indexAdded[kh] = struct{}{} + } + } + // Flush in chunks to keep pipeline buffers reasonable on heavy seeds. + if len(indexAdded)%500 == 0 { + if _, err := pipe.Exec(ctx); err != nil { + return err + } + pipe = s.rdb.Pipeline() + } + } + _, err := pipe.Exec(ctx) + return err +} + +// Write writes a single impression. K ZADDs + K idempotent SADDs, pipelined. +func (s *ZSetPerKeyedStore) Write(ctx context.Context, userID string, imp Impression) error { + keys := imp.FcapKeys + if len(keys) > MaxKeysPerImpression { + keys = keys[:MaxKeysPerImpression] + } + if len(keys) == 0 { + return nil + } + impHash := HashKey(imp.ImpressionID) + idx := zsetPerKeyedIndex(userID) + pipe := s.rdb.Pipeline() + for _, k := range keys { + kh := HashKey(k) + pipe.ZAdd(ctx, zsetPerKeyedKey(userID, kh), redis.Z{ + Score: float64(imp.Timestamp), + Member: encodePerKeyedMember(impHash), + }) + pipe.SAdd(ctx, idx, strconv.FormatUint(kh, 16)) + } + _, err := pipe.Exec(ctx) + return err +} + +// ReadAndCheck answers a single eligibility check. Single ZSET fetch. +func (s *ZSetPerKeyedStore) ReadAndCheck(ctx context.Context, userID string, fcapKeyHash uint64, rules []FrequencyRule, now int64) (capped bool, latestTS int64, err error) { + if len(rules) == 0 { + return false, 0, nil + } + maxWindow := rules[0].Window + for _, r := range rules[1:] { + if r.Window > maxWindow { + maxWindow = r.Window + } + } + cutoff := now - int64(maxWindow.Seconds()) + res, err := s.rdb.ZRangeByScoreWithScores(ctx, zsetPerKeyedKey(userID, fcapKeyHash), &redis.ZRangeBy{ + Min: strconv.FormatInt(cutoff, 10), + Max: "+inf", + }).Result() + if err != nil { + if errors.Is(err, redis.Nil) { + return false, 0, nil + } + return false, 0, err + } + type entry struct { + impHash uint64 + ts int64 + } + matched := make([]entry, 0, len(res)) + for _, z := range res { + impHash := decodePerKeyedMember(z.Member.(string)) + ts := int64(z.Score) + matched = append(matched, entry{impHash, ts}) + if ts > latestTS { + latestTS = ts + } + } + for _, rule := range rules { + ruleCutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for _, e := range matched { + if e.ts < ruleCutoff { + continue + } + if _, dup := seen[e.impHash]; dup { + continue + } + seen[e.impHash] = struct{}{} + count++ + } + if count >= rule.MaxCount { + capped = true + } + } + return capped, latestTS, nil +} + +// ReadBatchCheck answers eligibility for many fcap_keys. Pipelines a +// ZRANGEBYSCORE per key, processes results in parallel. +func (s *ZSetPerKeyedStore) ReadBatchCheck(ctx context.Context, userID string, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) (cappedByKey map[uint64]bool, err error) { + maxWindow := time.Duration(0) + for _, r := range rules { + if r.Window > maxWindow { + maxWindow = r.Window + } + } + cutoff := now - int64(maxWindow.Seconds()) + pipe := s.rdb.Pipeline() + cmds := make([]*redis.ZSliceCmd, len(fcapKeyHashes)) + for i, kh := range fcapKeyHashes { + cmds[i] = pipe.ZRangeByScoreWithScores(ctx, zsetPerKeyedKey(userID, kh), &redis.ZRangeBy{ + Min: strconv.FormatInt(cutoff, 10), + Max: "+inf", + }) + } + if _, err := pipe.Exec(ctx); err != nil && !errors.Is(err, redis.Nil) { + return nil, err + } + + cappedByKey = make(map[uint64]bool, len(fcapKeyHashes)) + for i, kh := range fcapKeyHashes { + res, err := cmds[i].Result() + if err != nil && !errors.Is(err, redis.Nil) { + return nil, err + } + capped := false + for _, rule := range rules { + ruleCutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for _, z := range res { + ts := int64(z.Score) + if ts < ruleCutoff { + continue + } + impHash := decodePerKeyedMember(z.Member.(string)) + if _, dup := seen[impHash]; dup { + continue + } + seen[impHash] = struct{}{} + count++ + } + if count >= rule.MaxCount { + capped = true + break + } + } + cappedByKey[kh] = capped + } + return cappedByKey, nil +} + +// Cleanup walks the per-user index and runs ZREMRANGEBYSCORE on each +// fcap_key ZSET. Empty ZSETs get their index entries removed. +func (s *ZSetPerKeyedStore) Cleanup(ctx context.Context, userID string, now int64, window time.Duration) error { + cutoff := now - int64(window.Seconds()) + idxKey := zsetPerKeyedIndex(userID) + members, err := s.rdb.SMembers(ctx, idxKey).Result() + if err != nil { + if errors.Is(err, redis.Nil) { + return nil + } + return err + } + pipe := s.rdb.Pipeline() + remCmds := make([]*redis.IntCmd, len(members)) + cardCmds := make([]*redis.IntCmd, len(members)) + for i, m := range members { + kh, err := strconv.ParseUint(m, 16, 64) + if err != nil { + continue + } + key := zsetPerKeyedKey(userID, kh) + remCmds[i] = pipe.ZRemRangeByScore(ctx, key, "-inf", "("+strconv.FormatInt(cutoff, 10)) + cardCmds[i] = pipe.ZCard(ctx, key) + } + if _, err := pipe.Exec(ctx); err != nil { + return err + } + // Second pass: SREM index entries for keys that are now empty. + pipe = s.rdb.Pipeline() + for i, m := range members { + if cardCmds[i] == nil { + continue + } + card, err := cardCmds[i].Result() + if err != nil { + continue + } + if card == 0 { + pipe.SRem(ctx, idxKey, m) + } + } + _, err = pipe.Exec(ctx) + return err +} + +// MemoryUsage sums valkey-reported memory for the index plus all per-key +// ZSETs the user has data in. +func (s *ZSetPerKeyedStore) MemoryUsage(ctx context.Context, userID string) (int64, error) { + idxKey := zsetPerKeyedIndex(userID) + members, err := s.rdb.SMembers(ctx, idxKey).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return 0, err + } + var total int64 + if u, err := s.rdb.MemoryUsage(ctx, idxKey).Result(); err == nil { + total += u + } + for _, m := range members { + kh, err := strconv.ParseUint(m, 16, 64) + if err != nil { + continue + } + if u, err := s.rdb.MemoryUsage(ctx, zsetPerKeyedKey(userID, kh)).Result(); err == nil { + total += u + } + } + return total, nil +} + +// Reset deletes the index plus all per-key ZSETs. +func (s *ZSetPerKeyedStore) Reset(ctx context.Context, userID string) error { + idxKey := zsetPerKeyedIndex(userID) + members, err := s.rdb.SMembers(ctx, idxKey).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return err + } + pipe := s.rdb.Pipeline() + pipe.Del(ctx, idxKey) + for _, m := range members { + kh, err := strconv.ParseUint(m, 16, 64) + if err != nil { + continue + } + pipe.Del(ctx, zsetPerKeyedKey(userID, kh)) + } + _, err = pipe.Exec(ctx) + return err +} + +// --- Split bench support --- + +// Fetch for the perkeyed variant: when a single fcap_key is in scope the +// fetch is a single ZRANGEBYSCORE; for batch reads the bench will call +// ReadBatchCheck directly (the pipelined fetch is part of that path). +// +// This Fetch implementation aggregates members across ALL fcap_keys in the +// user's index — useful for the split bench's "Process given pre-fetched +// data" model. In production, the win is that single-key reads skip this +// aggregation entirely. +func (s *ZSetPerKeyedStore) Fetch(ctx context.Context, userID string, window time.Duration, now int64) (FetchResult, error) { + cutoff := now - int64(window.Seconds()) + idxKey := zsetPerKeyedIndex(userID) + members, err := s.rdb.SMembers(ctx, idxKey).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return FetchResult{}, err + } + pipe := s.rdb.Pipeline() + cmds := make([]*redis.ZSliceCmd, 0, len(members)) + keyHashes := make([]uint64, 0, len(members)) + for _, m := range members { + kh, err := strconv.ParseUint(m, 16, 64) + if err != nil { + continue + } + keyHashes = append(keyHashes, kh) + cmds = append(cmds, pipe.ZRangeByScoreWithScores(ctx, zsetPerKeyedKey(userID, kh), &redis.ZRangeBy{ + Min: strconv.FormatInt(cutoff, 10), + Max: "+inf", + })) + } + if _, err := pipe.Exec(ctx); err != nil && !errors.Is(err, redis.Nil) { + return FetchResult{}, err + } + // Re-shape into FetchResult.ZMembers with a synthetic member encoding + // of {impHash, keyHash} so the existing Process step can decode it the + // same way as the per-key variant. + all := make([]redis.Z, 0) + for i, kh := range keyHashes { + zs, err := cmds[i].Result() + if err != nil && !errors.Is(err, redis.Nil) { + continue + } + for _, z := range zs { + impHash := decodePerKeyedMember(z.Member.(string)) + all = append(all, redis.Z{ + Score: z.Score, + Member: encodePerKeyMember(impHash, kh), + }) + } + } + return FetchResult{ZMembers: all}, nil +} + +// Process for the perkeyed variant: identical decoding to the per-key +// variant since Fetch reshapes into {impHash, keyHash} 16-byte members. +func (s *ZSetPerKeyedStore) Process(raw FetchResult, fcapKeyHashes []uint64, rules []FrequencyRule, now int64) map[uint64]bool { + wantedSet := make(map[uint64]struct{}, len(fcapKeyHashes)) + for _, k := range fcapKeyHashes { + wantedSet[k] = struct{}{} + } + type entry struct { + impHash uint64 + ts int64 + } + byKey := make(map[uint64][]entry, len(fcapKeyHashes)) + for _, z := range raw.ZMembers { + impHash, keyHash := decodePerKeyMember(z.Member.(string)) + if _, want := wantedSet[keyHash]; !want { + continue + } + byKey[keyHash] = append(byKey[keyHash], entry{impHash, int64(z.Score)}) + } + cappedByKey := make(map[uint64]bool, len(fcapKeyHashes)) + for _, kh := range fcapKeyHashes { + bucket := byKey[kh] + capped := false + for _, rule := range rules { + cutoff := now - int64(rule.Window.Seconds()) + seen := make(map[uint64]struct{}) + count := 0 + for _, e := range bucket { + if e.ts < cutoff { + continue + } + if _, dup := seen[e.impHash]; dup { + continue + } + seen[e.impHash] = struct{}{} + count++ + } + if count >= rule.MaxCount { + capped = true + break + } + } + cappedByKey[kh] = capped + } + return cappedByKey +}