1099 lines
34 KiB
Go
1099 lines
34 KiB
Go
package shortrecommend
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"reflect"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"91porn-server/common/redis"
|
|
"91porn-server/models/v/vidmod"
|
|
|
|
"go.mongodb.org/mongo-driver/bson/primitive"
|
|
)
|
|
|
|
type fakePublishClient struct {
|
|
queue []string
|
|
pipelineWindowSizes []int
|
|
pipelineBatchSizes []int
|
|
pipelineExpirations []time.Duration
|
|
pipelineCalls int
|
|
pipelineErrorAt int
|
|
pipelineDuplicateAt int
|
|
cancelAfterPipelineAt int
|
|
cancel context.CancelFunc
|
|
duplicateFirstPush bool
|
|
firstPushTTL int64
|
|
metaBuildingTTL int64
|
|
publishedTTL int64
|
|
metaLength int
|
|
swapAttempts int
|
|
swapped bool
|
|
deleteCalls int
|
|
queueLengthBeforeDelete int
|
|
swapScript string
|
|
swapKeys []string
|
|
swapArgs []interface{}
|
|
}
|
|
|
|
func (f *fakePublishClient) DelContext(_ context.Context, _ ...string) (int64, error) {
|
|
f.deleteCalls++
|
|
f.queueLengthBeforeDelete = len(f.queue)
|
|
f.queue = nil
|
|
return 1, nil
|
|
}
|
|
|
|
func (f *fakePublishClient) EvalContext(
|
|
_ context.Context,
|
|
script string,
|
|
keys []string,
|
|
args ...interface{},
|
|
) (interface{}, error) {
|
|
switch {
|
|
case strings.Contains(script, "for i = 1, #ARGV - 1"):
|
|
if len(args) < 2 {
|
|
return nil, fmt.Errorf("invalid first push args")
|
|
}
|
|
ids := make([]string, len(args)-1)
|
|
for i := range ids {
|
|
id, ok := args[i].(string)
|
|
if !ok {
|
|
return nil, fmt.Errorf("first push id %d is not a string", i)
|
|
}
|
|
ids[i] = id
|
|
}
|
|
ttl, ok := args[len(args)-1].(int64)
|
|
if !ok {
|
|
return nil, fmt.Errorf("first push ttl is not int64")
|
|
}
|
|
f.firstPushTTL = ttl
|
|
f.queue = append(f.queue, ids...)
|
|
if f.duplicateFirstPush {
|
|
f.queue = append(f.queue, ids...)
|
|
}
|
|
return int64(1), nil
|
|
case strings.Contains(script, "redis.call('HSET', KEYS[1]"):
|
|
length, ok := args[0].(int)
|
|
if !ok {
|
|
return nil, fmt.Errorf("meta length is not int")
|
|
}
|
|
f.metaLength = length
|
|
ttl, ok := args[3].(int64)
|
|
if !ok {
|
|
return nil, fmt.Errorf("meta ttl is not int64")
|
|
}
|
|
f.metaBuildingTTL = ttl
|
|
return int64(1), nil
|
|
case strings.Contains(script, "redis.call('RENAME'"):
|
|
f.swapAttempts++
|
|
f.swapScript = script
|
|
f.swapKeys = append([]string(nil), keys...)
|
|
f.swapArgs = append([]interface{}(nil), args...)
|
|
ttl, ok := args[1].(int64)
|
|
if !ok {
|
|
return nil, fmt.Errorf("published ttl is not int64")
|
|
}
|
|
f.publishedTTL = ttl
|
|
expected, ok := args[4].(int)
|
|
if !ok {
|
|
return nil, fmt.Errorf("expected queue length is not int")
|
|
}
|
|
if len(f.queue) != expected {
|
|
return int64(-2), nil
|
|
}
|
|
f.swapped = true
|
|
return int64(1), nil
|
|
default:
|
|
return nil, fmt.Errorf("unexpected script")
|
|
}
|
|
}
|
|
|
|
type fakeScriptClient struct {
|
|
result interface{}
|
|
err error
|
|
run func(*fakeScriptClient, context.Context, *redis.Script, []string, ...interface{}) (interface{}, error)
|
|
calls int
|
|
ctx context.Context
|
|
script *redis.Script
|
|
keys []string
|
|
args []interface{}
|
|
}
|
|
|
|
func (f *fakeScriptClient) RunScriptContext(
|
|
ctx context.Context,
|
|
script *redis.Script,
|
|
keys []string,
|
|
args ...interface{},
|
|
) (interface{}, error) {
|
|
f.calls++
|
|
f.ctx = ctx
|
|
f.script = script
|
|
f.keys = append([]string(nil), keys...)
|
|
f.args = append([]interface{}(nil), args...)
|
|
if f.run != nil {
|
|
return f.run(f, ctx, script, keys, args...)
|
|
}
|
|
return f.result, f.err
|
|
}
|
|
|
|
func (f *fakePublishClient) RPushPipelineContext(
|
|
_ context.Context,
|
|
_ string,
|
|
values []string,
|
|
batchSize int,
|
|
expirations ...time.Duration,
|
|
) (int64, error) {
|
|
f.pipelineCalls++
|
|
f.pipelineWindowSizes = append(f.pipelineWindowSizes, len(values))
|
|
f.pipelineBatchSizes = append(f.pipelineBatchSizes, batchSize)
|
|
if len(expirations) > 0 {
|
|
f.pipelineExpirations = append(f.pipelineExpirations, expirations[0])
|
|
}
|
|
if f.pipelineCalls == f.pipelineErrorAt {
|
|
f.queue = append(f.queue, values[:len(values)/2]...)
|
|
return 0, fmt.Errorf("injected pipeline error")
|
|
}
|
|
f.queue = append(f.queue, values...)
|
|
if f.pipelineCalls == f.pipelineDuplicateAt {
|
|
f.queue = append(f.queue, values...)
|
|
}
|
|
if f.pipelineCalls == f.cancelAfterPipelineAt && f.cancel != nil {
|
|
f.cancel()
|
|
}
|
|
return int64(len(f.queue)), nil
|
|
}
|
|
|
|
func candidate(n int, score int64, reviewAt time.Time) vidmod.RecommendCandidate {
|
|
return vidmod.RecommendCandidate{
|
|
ID: primitive.NewObjectIDFromTimestamp(time.Unix(int64(n+1), int64(n))),
|
|
RecommendScore: score,
|
|
ReviewAt: reviewAt,
|
|
}
|
|
}
|
|
|
|
func TestAssembleSeventeenHighThreeNew(t *testing.T) {
|
|
now := time.Date(2026, 7, 24, 8, 0, 0, 0, time.UTC)
|
|
input := make([]vidmod.RecommendCandidate, 0, 40)
|
|
for i := 0; i < 34; i++ {
|
|
input = append(input, candidate(i, int64(1000-i), now.Add(-48*time.Hour)))
|
|
}
|
|
for i := 0; i < 6; i++ {
|
|
input = append(input, candidate(100+i, 0, now.Add(-time.Duration(i)*time.Hour)))
|
|
}
|
|
got := Assemble(input, now)
|
|
if len(got) != 40 {
|
|
t.Fatalf("length=%d want=40", len(got))
|
|
}
|
|
newIDs := map[string]bool{}
|
|
for i := 0; i < 6; i++ {
|
|
newIDs[input[34+i].ID.Hex()] = true
|
|
}
|
|
for block := 0; block < 2; block++ {
|
|
count := 0
|
|
for _, id := range got[block*20 : block*20+20] {
|
|
if newIDs[id] {
|
|
count++
|
|
}
|
|
}
|
|
if count != 3 {
|
|
t.Fatalf("block %d new count=%d want=3", block, count)
|
|
}
|
|
}
|
|
assertUnique(t, got)
|
|
}
|
|
|
|
func TestAssembleWithStats(t *testing.T) {
|
|
now := time.Date(2026, 7, 24, 8, 0, 0, 0, time.UTC)
|
|
input := make([]vidmod.RecommendCandidate, 0, 23)
|
|
for i := 0; i < 20; i++ {
|
|
input = append(input, candidate(i, int64(1000-i), now.Add(-48*time.Hour)))
|
|
}
|
|
for i := 0; i < 3; i++ {
|
|
input = append(input, candidate(100+i, 0, now.Add(-time.Duration(i)*time.Hour)))
|
|
}
|
|
|
|
ids, stats := AssembleWithStats(input, now)
|
|
if len(ids) != len(input) {
|
|
t.Fatalf("len(ids) = %d, want %d", len(ids), len(input))
|
|
}
|
|
if stats.TotalCount != 23 || stats.HighCount != 20 ||
|
|
stats.NewCount != 3 || stats.BlockCount != 2 {
|
|
t.Fatalf("stats = %+v", stats)
|
|
}
|
|
}
|
|
|
|
func TestAssembleNewShortageFilledByHigh(t *testing.T) {
|
|
now := time.Now()
|
|
for newCount := 0; newCount <= 3; newCount++ {
|
|
t.Run(fmt.Sprintf("new_%d", newCount), func(t *testing.T) {
|
|
input := make([]vidmod.RecommendCandidate, 0, 20)
|
|
for i := 0; i < 20-newCount; i++ {
|
|
input = append(input, candidate(i, int64(100-i), now.Add(-48*time.Hour)))
|
|
}
|
|
for i := 0; i < newCount; i++ {
|
|
input = append(input, candidate(100+i, 0, now.Add(-time.Hour)))
|
|
}
|
|
got := Assemble(input, now)
|
|
if len(got) != 20 {
|
|
t.Fatalf("length=%d want=20", len(got))
|
|
}
|
|
assertUnique(t, got)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestAssembleTailAndDeterminism(t *testing.T) {
|
|
now := time.Date(2026, 7, 24, 8, 0, 0, 0, time.UTC)
|
|
input := make([]vidmod.RecommendCandidate, 0, 9)
|
|
for i := 0; i < 7; i++ {
|
|
input = append(input, candidate(i, int64(i), now.Add(-48*time.Hour)))
|
|
}
|
|
input = append(input, candidate(100, 0, now.Add(-time.Hour)), candidate(101, 0, now.Add(-2*time.Hour)))
|
|
first, second := Assemble(input, now), Assemble(input, now)
|
|
if fmt.Sprint(first) != fmt.Sprint(second) {
|
|
t.Fatalf("same version must be deterministic")
|
|
}
|
|
if len(first) != 9 {
|
|
t.Fatalf("tail length=%d want=9", len(first))
|
|
}
|
|
assertUnique(t, first)
|
|
}
|
|
|
|
func TestAssembleReviewAtBoundariesExcludesFuture(t *testing.T) {
|
|
now := time.Date(2026, 7, 24, 8, 0, 0, 0, time.UTC)
|
|
input := []vidmod.RecommendCandidate{
|
|
candidate(1, 100, now.Add(-24*time.Hour)),
|
|
candidate(2, 99, now.Add(-24*time.Hour-time.Nanosecond)),
|
|
candidate(3, 98, now.Add(time.Nanosecond)),
|
|
}
|
|
got, stats := AssembleWithStats(input, now)
|
|
if len(got) != 2 {
|
|
t.Fatalf("length=%d want=2", len(got))
|
|
}
|
|
assertUnique(t, got)
|
|
if stats.TotalCount != 2 || stats.NewCount != 1 || stats.HighCount != 1 {
|
|
t.Fatalf("stats=%+v want total=2 new=1 high=1", stats)
|
|
}
|
|
futureID := input[2].ID.Hex()
|
|
for _, id := range got {
|
|
if id == futureID {
|
|
t.Fatalf("future reviewAt video %s must not enter recommendation pools", futureID)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestTakeRejectsNonPositiveSizeBeforeRedis(t *testing.T) {
|
|
for _, size := range []int{0, -1} {
|
|
result, err := Take(nil, 123, size)
|
|
if err != nil {
|
|
t.Fatalf("Take(size=%d) error = %v", size, err)
|
|
}
|
|
if result.Length != 0 || len(result.IDs) != 0 {
|
|
t.Fatalf("Take(size=%d) result = %+v, want empty", size, result)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestTakeVersionContextUsesReusableScriptAndExpectedVersion(t *testing.T) {
|
|
ctx := context.WithValue(context.Background(), struct{}{}, "request")
|
|
client := &fakeScriptClient{
|
|
result: []interface{}{"20260731", "4", "2", "video-3", "video-4"},
|
|
}
|
|
result, err := takeContext(ctx, client, 123, 2, "20260731", "")
|
|
if err != nil {
|
|
t.Fatalf("TakeVersionContext() error = %v", err)
|
|
}
|
|
if client.calls != 1 || client.ctx != ctx {
|
|
t.Fatalf("script calls = %d context propagated = %v", client.calls, client.ctx == ctx)
|
|
}
|
|
if client.script != takeRedisScript || client.script.Hash() == "" {
|
|
t.Fatal("TakeVersionContext() did not reuse the cached Redis script")
|
|
}
|
|
if got := client.args[5]; got != "20260731" {
|
|
t.Fatalf("expected version arg = %v, want 20260731", got)
|
|
}
|
|
if result.Version != "20260731" || result.Length != 4 || result.Offset != 2 ||
|
|
!reflect.DeepEqual(result.IDs, []string{"video-3", "video-4"}) {
|
|
t.Fatalf("TakeVersionContext() result = %+v", result)
|
|
}
|
|
}
|
|
|
|
func TestTakeVersionContextReportsVersionChangeWithoutIDs(t *testing.T) {
|
|
client := &fakeScriptClient{
|
|
result: []interface{}{"20260801", "-1", "0"},
|
|
}
|
|
result, err := takeContext(
|
|
context.Background(), client, 123, 20, "20260731", "",
|
|
)
|
|
if !errors.Is(err, ErrVersionChanged) {
|
|
t.Fatalf("TakeVersionContext() error = %v, want ErrVersionChanged", err)
|
|
}
|
|
if result.Version != "20260801" || len(result.IDs) != 0 {
|
|
t.Fatalf("TakeVersionContext() result = %+v", result)
|
|
}
|
|
}
|
|
|
|
func TestTakeVersionContextReportsUnhealthyQueue(t *testing.T) {
|
|
client := &fakeScriptClient{
|
|
result: []interface{}{"20260731", "-2", "0"},
|
|
}
|
|
_, err := takeContext(context.Background(), client, 123, 20, "20260731", "")
|
|
if !errors.Is(err, ErrQueueUnhealthy) {
|
|
t.Fatalf("TakeVersionContext() error = %v, want ErrQueueUnhealthy", err)
|
|
}
|
|
}
|
|
|
|
func TestTakeContextStopsBeforeRedisWhenCancelled(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
client := &fakeScriptClient{}
|
|
_, err := takeContext(ctx, client, 123, 20, "20260731", "")
|
|
if !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("TakeVersionContext() error = %v, want context.Canceled", err)
|
|
}
|
|
if client.calls != 0 {
|
|
t.Fatalf("script calls = %d, want 0", client.calls)
|
|
}
|
|
}
|
|
|
|
func TestTakeIdempotentContextBuildsStableScopedCacheKey(t *testing.T) {
|
|
client := &fakeScriptClient{
|
|
result: []interface{}{"20260731", "4", "0", "video-1", "video-2"},
|
|
}
|
|
_, err := takeIdempotentContext(
|
|
context.Background(), client, 123, 2, "20260731", " request-1 ", 3,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("takeIdempotentContext() error = %v", err)
|
|
}
|
|
idempotencyKey, ok := client.args[6].(string)
|
|
if !ok || !strings.HasPrefix(
|
|
idempotencyKey,
|
|
"recommend:short:take-cache:123:",
|
|
) || !strings.HasSuffix(idempotencyKey, ":3") {
|
|
t.Fatalf("idempotency key = %#v", client.args[6])
|
|
}
|
|
if strings.Contains(idempotencyKey, "request-1") {
|
|
t.Fatal("raw request ID must not be copied into Redis keys")
|
|
}
|
|
if client.args[7] != int64(takeIdempotencyTTL/time.Second) {
|
|
t.Fatalf("idempotency ttl = %v", client.args[7])
|
|
}
|
|
}
|
|
|
|
func TestTakeIdempotentContextAllowsFirstBatchWithoutExpectedVersion(t *testing.T) {
|
|
client := &fakeScriptClient{
|
|
result: []interface{}{"20260731", "4", "0", "video-1"},
|
|
}
|
|
result, err := takeIdempotentContext(
|
|
context.Background(), client, 123, 1, "", "request-1", 0,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("takeIdempotentContext() error = %v", err)
|
|
}
|
|
if result.Version != "20260731" || client.args[5] != "" {
|
|
t.Fatalf("result = %+v expected version arg = %#v", result, client.args[5])
|
|
}
|
|
}
|
|
|
|
func TestTakeCurrentIdempotencyCacheIsScopedByLuaCurrentVersion(t *testing.T) {
|
|
if !strings.Contains(
|
|
takeScriptSource,
|
|
"idempotencyKey = ARGV[7] .. ':' .. version",
|
|
) {
|
|
t.Fatal("take script must scope its idempotency cache by the atomic current version")
|
|
}
|
|
first := &fakeScriptClient{
|
|
result: []interface{}{"20260731", "4", "0", "video-1"},
|
|
}
|
|
second := &fakeScriptClient{
|
|
result: []interface{}{"20260801", "4", "0", "video-2"},
|
|
}
|
|
if _, err := takeIdempotentContext(
|
|
context.Background(), first, 123, 1, "", "request-1", 0,
|
|
); err != nil {
|
|
t.Fatalf("first take error = %v", err)
|
|
}
|
|
if _, err := takeIdempotentContext(
|
|
context.Background(), second, 123, 1, "", "request-1", 0,
|
|
); err != nil {
|
|
t.Fatalf("second take error = %v", err)
|
|
}
|
|
if first.args[6] != second.args[6] {
|
|
t.Fatalf("cache base differs: first=%v second=%v", first.args[6], second.args[6])
|
|
}
|
|
// Lua appends the version it read atomically, yielding different physical
|
|
// keys "<base>:20260731" and "<base>:20260801".
|
|
}
|
|
|
|
func TestTakeIdempotencyCacheStoresReservationInsteadOfVideoIDs(t *testing.T) {
|
|
for _, want := range []string{
|
|
"local function readBatch",
|
|
"local cachedWanted = tonumber(cached[4])",
|
|
"version, tostring(length), tostring(offset), tostring(wanted)",
|
|
} {
|
|
if !strings.Contains(takeScriptSource, want) {
|
|
t.Fatalf("take script missing compact reservation logic %q", want)
|
|
}
|
|
}
|
|
if strings.Contains(takeScriptSource, "unpack(result)") {
|
|
t.Fatal("take script must not duplicate all video IDs in idempotency cache")
|
|
}
|
|
}
|
|
|
|
func TestTakeIdempotentContextValidatesScope(t *testing.T) {
|
|
longRequestID := strings.Repeat("x", maxTakeRequestIDLength+1)
|
|
tests := []struct {
|
|
name string
|
|
expectedVersion string
|
|
requestID string
|
|
batchIndex int
|
|
}{
|
|
{name: "blank request", expectedVersion: "20260731", requestID: " ", batchIndex: 0},
|
|
{name: "long request", expectedVersion: "20260731", requestID: longRequestID, batchIndex: 0},
|
|
{name: "negative batch", expectedVersion: "20260731", requestID: "request", batchIndex: -1},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
client := &fakeScriptClient{}
|
|
_, err := takeIdempotentContext(
|
|
context.Background(), client, 1, 20,
|
|
tt.expectedVersion, tt.requestID, tt.batchIndex,
|
|
)
|
|
if err == nil {
|
|
t.Fatal("takeIdempotentContext() should reject invalid scope")
|
|
}
|
|
if client.calls != 0 {
|
|
t.Fatalf("script calls = %d, want 0", client.calls)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestTakeScriptUsesFixedAbsoluteOffsetExpiration(t *testing.T) {
|
|
if !strings.Contains(takeScriptSource, "redis.call('PEXPIREAT', offsetKey, expiresAtMs)") {
|
|
t.Fatal("take script must align offset to the queue absolute expiration")
|
|
}
|
|
if !strings.Contains(takeScriptSource, "if offsetKeyTTL <= 0 then") {
|
|
t.Fatal("take script must set offset expiration only when the hash has no TTL")
|
|
}
|
|
if strings.Contains(takeScriptSource, "redis.call('EXPIRE', offsetKey") {
|
|
t.Fatal("take script must not renew a sliding offset TTL")
|
|
}
|
|
}
|
|
|
|
func TestReserveContextUsesStableReceiptAndRandomLeaseWithoutAdvancingOffset(t *testing.T) {
|
|
newClient := func() *fakeScriptClient {
|
|
return &fakeScriptClient{
|
|
run: func(
|
|
f *fakeScriptClient,
|
|
_ context.Context,
|
|
_ *redis.Script,
|
|
_ []string,
|
|
args ...interface{},
|
|
) (interface{}, error) {
|
|
token := fmt.Sprint(args[8])
|
|
return []interface{}{
|
|
"20260731-build", "RESERVED", "4", "3", "2", token,
|
|
"video-4", "video-1",
|
|
}, nil
|
|
},
|
|
}
|
|
}
|
|
firstClient, secondClient := newClient(), newClient()
|
|
first, err := reserveContext(
|
|
context.Background(), firstClient, 123, 2, "", "request-1",
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("first reserve error = %v", err)
|
|
}
|
|
second, err := reserveContext(
|
|
context.Background(), secondClient, 123, 2, "", "request-1",
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("second reserve error = %v", err)
|
|
}
|
|
if first.Version != "20260731-build" || first.Offset != 3 ||
|
|
first.Reserved != 2 ||
|
|
!reflect.DeepEqual(first.IDs, []string{"video-4", "video-1"}) {
|
|
t.Fatalf("first reservation = %+v", first)
|
|
}
|
|
if first.ReceiptID == "" || first.ReceiptID != second.ReceiptID {
|
|
t.Fatalf("receipt IDs first=%q second=%q", first.ReceiptID, second.ReceiptID)
|
|
}
|
|
if first.LeaseToken == "" || first.LeaseToken == second.LeaseToken {
|
|
t.Fatalf("lease tokens first=%q second=%q", first.LeaseToken, second.LeaseToken)
|
|
}
|
|
if firstClient.script != reserveRedisScript ||
|
|
firstClient.args[10] != int64(reservationTTL/time.Millisecond) {
|
|
t.Fatal("reserve did not use the reusable script and bounded lease TTL")
|
|
}
|
|
if strings.Contains(reserveScriptSource, "HSET', offsetKey") {
|
|
t.Fatal("Reserve must never advance or create the user offset hash")
|
|
}
|
|
for _, want := range []string{
|
|
"'HMGET', leaseKey",
|
|
"redis.call('PEXPIRE', leaseKey",
|
|
"'COMMITTED'",
|
|
"'BUSY'",
|
|
} {
|
|
if !strings.Contains(reserveScriptSource, want) {
|
|
t.Fatalf("reserve script missing %q", want)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestReserveContextParsesCommittedReceiptAndErrors(t *testing.T) {
|
|
committedClient := &fakeScriptClient{
|
|
result: []interface{}{
|
|
"20260731-old", "COMMITTED", "4", "1", "2", "lease",
|
|
"video-2", "video-3",
|
|
},
|
|
}
|
|
committed, err := reserveContext(
|
|
context.Background(), committedClient, 123, 2, "", "request",
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("committed reserve error = %v", err)
|
|
}
|
|
if !committed.AlreadyCommitted || committed.Reserved != 2 {
|
|
t.Fatalf("committed reservation = %+v", committed)
|
|
}
|
|
|
|
tests := []struct {
|
|
status string
|
|
want error
|
|
}{
|
|
{status: "VERSION_CHANGED", want: ErrVersionChanged},
|
|
{status: "UNHEALTHY", want: ErrQueueUnhealthy},
|
|
{status: "BUSY", want: ErrReservationBusy},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.status, func(t *testing.T) {
|
|
client := &fakeScriptClient{
|
|
result: []interface{}{"20260731", tt.status, "0", "0", "0", ""},
|
|
}
|
|
result, err := reserveContext(
|
|
context.Background(), client, 1, 20, "", "request",
|
|
)
|
|
if !errors.Is(err, tt.want) || result.Version != "20260731" {
|
|
t.Fatalf("result=%+v error=%v want=%v", result, err, tt.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestReserveContextRetriesUncertainResponseWithSameLease(t *testing.T) {
|
|
wantErr := errors.New("connection reset after write")
|
|
client := &fakeScriptClient{}
|
|
var firstArgs []interface{}
|
|
client.run = func(
|
|
f *fakeScriptClient,
|
|
_ context.Context,
|
|
_ *redis.Script,
|
|
_ []string,
|
|
args ...interface{},
|
|
) (interface{}, error) {
|
|
if f.calls == 1 {
|
|
firstArgs = append([]interface{}(nil), args...)
|
|
return nil, wantErr
|
|
}
|
|
if !reflect.DeepEqual(args, firstArgs) {
|
|
t.Fatalf("retry args changed:\nfirst=%#v\nretry=%#v", firstArgs, args)
|
|
}
|
|
return []interface{}{
|
|
"20260731-build", "RESERVED", "2", "0", "2",
|
|
fmt.Sprint(args[8]), "video-1", "video-2",
|
|
}, nil
|
|
}
|
|
|
|
reservation, err := reserveContext(
|
|
context.Background(), client, 123, 2, "", "request",
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("reserveContext() error = %v", err)
|
|
}
|
|
if client.calls != 2 || reservation.LeaseToken != fmt.Sprint(firstArgs[8]) ||
|
|
reservation.ReceiptID != fmt.Sprint(firstArgs[9]) {
|
|
t.Fatalf("calls=%d reservation=%+v args=%#v", client.calls, reservation, firstArgs)
|
|
}
|
|
}
|
|
|
|
func TestCommitReservationContextUsesCASReceiptAndAbsoluteTTL(t *testing.T) {
|
|
reservation := Reservation{
|
|
Version: "20260731-build",
|
|
Length: 10,
|
|
Offset: 8,
|
|
Reserved: 5,
|
|
LeaseToken: "lease",
|
|
ReceiptID: "receipt",
|
|
}
|
|
for _, raw := range []interface{}{int64(1), int64(2)} {
|
|
client := &fakeScriptClient{result: raw}
|
|
if err := commitReservationContext(
|
|
context.Background(), client, 123, reservation, 3,
|
|
); err != nil {
|
|
t.Fatalf("commit result %v error = %v", raw, err)
|
|
}
|
|
if client.script != commitReservationRedisScript {
|
|
t.Fatal("commit did not reuse cached Redis script")
|
|
}
|
|
if client.args[11] != 3 ||
|
|
client.args[12] != int64(takeIdempotencyTTL/time.Second) ||
|
|
client.args[13] != "recommend:short:commit-receipt:" {
|
|
t.Fatalf("commit args = %#v", client.args)
|
|
}
|
|
}
|
|
for _, want := range []string{
|
|
"currentOffset ~= tonumber(lease[4])",
|
|
"(currentOffset + consumed) % length",
|
|
"redis.call('PEXPIREAT', offsetKey, expiresAtMs)",
|
|
"'RPUSH', receiptKey",
|
|
} {
|
|
if !strings.Contains(commitReservationScriptSource, want) {
|
|
t.Fatalf("commit script missing %q", want)
|
|
}
|
|
}
|
|
if strings.Index(commitReservationScriptSource, "local receipt =") >
|
|
strings.Index(commitReservationScriptSource, "local version =") {
|
|
t.Fatal("commit must check its receipt before current for uncertain-response retries")
|
|
}
|
|
}
|
|
|
|
func TestCommitReservationContextMapsRejectedStates(t *testing.T) {
|
|
reservation := Reservation{
|
|
Version: "20260731-build",
|
|
Length: 10,
|
|
Offset: 0,
|
|
Reserved: 5,
|
|
LeaseToken: "lease",
|
|
ReceiptID: "receipt",
|
|
}
|
|
tests := []struct {
|
|
raw int64
|
|
want error
|
|
}{
|
|
{raw: -1, want: ErrVersionChanged},
|
|
{raw: -2, want: ErrQueueUnhealthy},
|
|
{raw: -3, want: ErrReservationExpired},
|
|
{raw: -4, want: ErrReservationConflict},
|
|
{raw: -5, want: ErrReservationConflict},
|
|
}
|
|
for _, tt := range tests {
|
|
client := &fakeScriptClient{result: tt.raw}
|
|
err := commitReservationContext(
|
|
context.Background(), client, 123, reservation, 3,
|
|
)
|
|
if !errors.Is(err, tt.want) {
|
|
t.Fatalf("raw=%d error=%v want=%v", tt.raw, err, tt.want)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestAbortReservationContextIsTokenScoped(t *testing.T) {
|
|
client := &fakeScriptClient{result: int64(1)}
|
|
reservation := Reservation{
|
|
Version: "20260731-build",
|
|
LeaseToken: "lease",
|
|
ReceiptID: "receipt",
|
|
}
|
|
if err := abortReservationContext(
|
|
context.Background(), client, 123, reservation,
|
|
); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if client.script != abortReservationRedisScript ||
|
|
!reflect.DeepEqual(client.args, []interface{}{
|
|
"recommend:short:reservation:",
|
|
"20260731-build",
|
|
"123",
|
|
"lease",
|
|
"receipt",
|
|
}) {
|
|
t.Fatalf("abort script=%p args=%#v", client.script, client.args)
|
|
}
|
|
if !strings.Contains(abortReservationScriptSource, "lease[1] ~= ARGV[4]") ||
|
|
!strings.Contains(abortReservationScriptSource, "lease[2] ~= ARGV[5]") {
|
|
t.Fatal("abort must not delete a newer lease after TTL takeover")
|
|
}
|
|
}
|
|
|
|
func TestReservationAPIsRejectInvalidInputsBeforeRedis(t *testing.T) {
|
|
client := &fakeScriptClient{}
|
|
if _, err := reserveContext(
|
|
context.Background(), client, 0, 20, "", "",
|
|
); err == nil {
|
|
t.Fatal("reserve should reject uid 0")
|
|
}
|
|
invalid := Reservation{
|
|
Version: "20260731",
|
|
Length: 10,
|
|
Offset: 0,
|
|
Reserved: 2,
|
|
LeaseToken: "lease",
|
|
ReceiptID: "receipt",
|
|
}
|
|
if err := commitReservationContext(
|
|
context.Background(), client, 1, invalid, 3,
|
|
); err == nil {
|
|
t.Fatal("commit should reject consumed > reserved")
|
|
}
|
|
if client.calls != 0 {
|
|
t.Fatalf("Redis calls = %d, want 0", client.calls)
|
|
}
|
|
}
|
|
|
|
func TestHealthContextParsesAtomicHealthResult(t *testing.T) {
|
|
client := &fakeScriptClient{
|
|
result: []interface{}{"OK", "20260731", "154290", "154290", "7200000", "7199999"},
|
|
}
|
|
health, err := healthContext(context.Background(), client, "20260731")
|
|
if err != nil {
|
|
t.Fatalf("HealthContext() error = %v", err)
|
|
}
|
|
if !health.Healthy || health.Status != "OK" || health.Version != "20260731" ||
|
|
health.QueueLength != 154290 || health.MetadataLength != 154290 ||
|
|
health.QueueTTL != 2*time.Hour ||
|
|
health.MetadataTTL != 2*time.Hour-time.Millisecond {
|
|
t.Fatalf("HealthContext() = %+v", health)
|
|
}
|
|
if client.script != healthRedisScript || client.args[0] != "20260731" {
|
|
t.Fatal("HealthContext() did not use the expected cached script/version")
|
|
}
|
|
}
|
|
|
|
func TestHealthContextReturnsUnhealthyStatusWithoutRedisError(t *testing.T) {
|
|
client := &fakeScriptClient{
|
|
result: []interface{}{"LENGTH_MISMATCH", "20260731", "20", "19", "1000", "1000"},
|
|
}
|
|
health, err := healthContext(context.Background(), client, "20260731")
|
|
if err != nil {
|
|
t.Fatalf("HealthContext() error = %v", err)
|
|
}
|
|
if health.Healthy || health.Status != "LENGTH_MISMATCH" {
|
|
t.Fatalf("HealthContext() = %+v", health)
|
|
}
|
|
}
|
|
|
|
func TestTakeAndHealthRejectMalformedScriptResults(t *testing.T) {
|
|
takeClient := &fakeScriptClient{
|
|
result: []interface{}{"20260731", "not-a-length", "0"},
|
|
}
|
|
if _, err := takeContext(
|
|
context.Background(), takeClient, 1, 20, "", "",
|
|
); err == nil {
|
|
t.Fatal("takeContext() should reject malformed length")
|
|
}
|
|
healthClient := &fakeScriptClient{
|
|
result: []interface{}{"OK", "20260731", "20", "20", "not-a-ttl", "1000"},
|
|
}
|
|
if _, err := healthContext(context.Background(), healthClient, "20260731"); err == nil {
|
|
t.Fatal("HealthContext() should reject malformed TTL")
|
|
}
|
|
}
|
|
|
|
func TestSetKeyTTLHoursValidatesAndPublishesConfiguredTTL(t *testing.T) {
|
|
originalHours := int(KeyTTL() / time.Hour)
|
|
defer func() {
|
|
if err := SetKeyTTLHours(originalHours); err != nil {
|
|
t.Fatalf("restore key TTL: %v", err)
|
|
}
|
|
}()
|
|
for _, hours := range []int{minKeyTTLHours - 1, maxKeyTTLHours + 1} {
|
|
if err := SetKeyTTLHours(hours); err == nil {
|
|
t.Fatalf("SetKeyTTLHours(%d) should fail", hours)
|
|
}
|
|
}
|
|
const configuredHours = 96
|
|
if err := SetKeyTTLHours(configuredHours); err != nil {
|
|
t.Fatalf("SetKeyTTLHours() error = %v", err)
|
|
}
|
|
client := &fakePublishClient{}
|
|
if err := publish(
|
|
context.Background(), client, publishTestMeta(1), []string{"video-1"}, "lock", "token",
|
|
); err != nil {
|
|
t.Fatalf("publish() error = %v", err)
|
|
}
|
|
if client.publishedTTL != int64(96*time.Hour/time.Second) {
|
|
t.Fatalf("published ttl = %d, want %d", client.publishedTTL, int64(96*time.Hour/time.Second))
|
|
}
|
|
}
|
|
|
|
func TestPublishRejectsVersionMismatchBeforeRedis(t *testing.T) {
|
|
meta := QueueMeta{
|
|
Version: "20260730",
|
|
GeneratedAt: time.Date(2026, 7, 31, 0, 0, 0, 0, time.UTC),
|
|
}
|
|
err := Publish(context.Background(), nil, meta, []string{"video-id"}, "lock", "token")
|
|
if err == nil {
|
|
t.Fatal("Publish() should reject a version that does not match generatedAt")
|
|
}
|
|
}
|
|
|
|
func TestVersionWithRevisionKeepsDateAndSeparatesSameDayBuilds(t *testing.T) {
|
|
generatedAt := time.Date(2026, 7, 31, 2, 0, 0, 0, time.UTC)
|
|
first := VersionWithRevision(generatedAt, "build-a")
|
|
second := VersionWithRevision(generatedAt, "build-b")
|
|
if first != "20260731-build-a" || second != "20260731-build-b" {
|
|
t.Fatalf("versions = %q, %q", first, second)
|
|
}
|
|
if first == second {
|
|
t.Fatal("same-day builds must use different queue versions")
|
|
}
|
|
for _, version := range []string{"20260731", first, second} {
|
|
if !versionMatchesDate(version, "20260731") {
|
|
t.Fatalf("version %q should belong to date 20260731", version)
|
|
}
|
|
}
|
|
for _, version := range []string{
|
|
"20260730-build-a",
|
|
"202607310-build-a",
|
|
"20260731-",
|
|
} {
|
|
if versionMatchesDate(version, "20260731") {
|
|
t.Fatalf("version %q must not belong to date 20260731", version)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestPublishAcceptsSameDayRevisionVersion(t *testing.T) {
|
|
meta := publishTestMeta(1)
|
|
meta.Version = VersionWithRevision(meta.GeneratedAt, "build-a")
|
|
client := &fakePublishClient{}
|
|
|
|
if err := publish(
|
|
context.Background(), client, meta, []string{"video-1"}, "lock", "token",
|
|
); err != nil {
|
|
t.Fatalf("publish() error = %v", err)
|
|
}
|
|
if !client.swapped || client.swapArgs[0] != meta.Version {
|
|
t.Fatalf("same-day revision version was not published: %#v", client.swapArgs)
|
|
}
|
|
}
|
|
|
|
func TestHealthScriptAcceptsPureDateAndSameDayRevisionOnly(t *testing.T) {
|
|
for _, want := range []string{
|
|
"version == expectedDate",
|
|
"string.len(version) > string.len(expectedDate) + 1",
|
|
"expectedDate .. '-'",
|
|
} {
|
|
if !strings.Contains(healthScriptSource, want) {
|
|
t.Fatalf("health script missing date-family check %q", want)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestPublishUsesBoundedPipelineAndPreservesOrder(t *testing.T) {
|
|
total := queuePushBatchSize + queuePushBatchSize*queuePipelineBatchCount + 57
|
|
ids := make([]string, total)
|
|
for i := range ids {
|
|
ids[i] = fmt.Sprintf("video-%06d", i)
|
|
}
|
|
meta := publishTestMeta(total)
|
|
client := &fakePublishClient{}
|
|
|
|
if err := publish(context.Background(), client, meta, ids, "lock", "token"); err != nil {
|
|
t.Fatalf("publish() error = %v", err)
|
|
}
|
|
if !client.swapped || client.swapAttempts != 1 {
|
|
t.Fatalf("swap state = %v, attempts = %d, want successful single swap", client.swapped, client.swapAttempts)
|
|
}
|
|
if !reflect.DeepEqual(client.queue, ids) {
|
|
t.Fatal("published queue order differs from input")
|
|
}
|
|
wantWindows := []int{queuePushBatchSize * queuePipelineBatchCount, 57}
|
|
if !reflect.DeepEqual(client.pipelineWindowSizes, wantWindows) {
|
|
t.Fatalf("pipeline windows = %v, want %v", client.pipelineWindowSizes, wantWindows)
|
|
}
|
|
for _, batchSize := range client.pipelineBatchSizes {
|
|
if batchSize != queuePushBatchSize {
|
|
t.Fatalf("pipeline batch size = %d, want %d", batchSize, queuePushBatchSize)
|
|
}
|
|
}
|
|
if len(client.pipelineExpirations) != len(wantWindows) {
|
|
t.Fatalf("pipeline expirations = %v, want one per window", client.pipelineExpirations)
|
|
}
|
|
for _, expiration := range client.pipelineExpirations {
|
|
if expiration != buildingKeyTTL {
|
|
t.Fatalf("pipeline expiration = %v, want %v", expiration, buildingKeyTTL)
|
|
}
|
|
}
|
|
if client.firstPushTTL != int64(buildingKeyTTL/time.Second) {
|
|
t.Fatalf("building ttl = %d, want %d", client.firstPushTTL, int64(buildingKeyTTL/time.Second))
|
|
}
|
|
if client.metaBuildingTTL != int64(buildingKeyTTL/time.Second) {
|
|
t.Fatalf("meta building ttl = %d, want %d", client.metaBuildingTTL, int64(buildingKeyTTL/time.Second))
|
|
}
|
|
if client.publishedTTL != int64(KeyTTL()/time.Second) {
|
|
t.Fatalf("published ttl = %d, want %d", client.publishedTTL, int64(KeyTTL()/time.Second))
|
|
}
|
|
if client.metaLength != total {
|
|
t.Fatalf("meta length = %d, want %d", client.metaLength, total)
|
|
}
|
|
if !strings.Contains(client.swapScript, "redis.call('UNLINK'") {
|
|
t.Fatal("swap script must asynchronously remove current and previous offset hashes")
|
|
}
|
|
if !strings.Contains(client.swapScript, "redis.call('PEXPIREAT'") {
|
|
t.Fatal("swap script must set one absolute expiration for queue and metadata")
|
|
}
|
|
if len(client.swapArgs) != 6 || client.swapArgs[5] != "recommend:short:offset:" {
|
|
t.Fatalf("swap args = %#v, want previous offset prefix", client.swapArgs)
|
|
}
|
|
}
|
|
|
|
func TestPublishPipelineFailureCleansBuildingQueueWithoutSwap(t *testing.T) {
|
|
ids := make([]string, queuePushBatchSize+101)
|
|
for i := range ids {
|
|
ids[i] = fmt.Sprintf("video-%d", i)
|
|
}
|
|
client := &fakePublishClient{pipelineErrorAt: 1}
|
|
|
|
err := publish(context.Background(), client, publishTestMeta(len(ids)), ids, "lock", "token")
|
|
if err == nil || !strings.Contains(err.Error(), "injected pipeline error") {
|
|
t.Fatalf("publish() error = %v, want injected pipeline error", err)
|
|
}
|
|
if client.swapped || client.swapAttempts != 0 {
|
|
t.Fatalf("swap state = %v, attempts = %d, want no swap", client.swapped, client.swapAttempts)
|
|
}
|
|
if client.deleteCalls == 0 || client.queueLengthBeforeDelete <= queuePushBatchSize {
|
|
t.Fatalf(
|
|
"cleanup calls = %d, pre-delete length = %d, want partial queue cleanup",
|
|
client.deleteCalls, client.queueLengthBeforeDelete,
|
|
)
|
|
}
|
|
}
|
|
|
|
func TestPublishRejectsPipelineDuplicateBeforeSwap(t *testing.T) {
|
|
ids := make([]string, queuePushBatchSize+10)
|
|
for i := range ids {
|
|
ids[i] = fmt.Sprintf("video-%d", i)
|
|
}
|
|
client := &fakePublishClient{pipelineDuplicateAt: 1}
|
|
|
|
err := publish(context.Background(), client, publishTestMeta(len(ids)), ids, "lock", "token")
|
|
if err == nil || !strings.Contains(err.Error(), "queue length mismatch after pipeline") {
|
|
t.Fatalf("publish() error = %v, want pipeline length mismatch", err)
|
|
}
|
|
if client.swapped || client.swapAttempts != 0 {
|
|
t.Fatalf("swap state = %v, attempts = %d, want no swap", client.swapped, client.swapAttempts)
|
|
}
|
|
if client.deleteCalls == 0 {
|
|
t.Fatal("duplicate pipeline queue was not cleaned")
|
|
}
|
|
}
|
|
|
|
func TestPublishFinalLengthGateRejectsFirstPushDuplicate(t *testing.T) {
|
|
ids := []string{"video-1", "video-2"}
|
|
client := &fakePublishClient{duplicateFirstPush: true}
|
|
|
|
err := publish(context.Background(), client, publishTestMeta(len(ids)), ids, "lock", "token")
|
|
if err == nil || !strings.Contains(err.Error(), "queue length mismatch before queue swap") {
|
|
t.Fatalf("publish() error = %v, want final queue length mismatch", err)
|
|
}
|
|
if client.swapped || client.swapAttempts != 1 {
|
|
t.Fatalf("swap state = %v, attempts = %d, want rejected swap", client.swapped, client.swapAttempts)
|
|
}
|
|
if client.deleteCalls == 0 {
|
|
t.Fatal("invalid first push queue was not cleaned")
|
|
}
|
|
}
|
|
|
|
func TestPublishCancellationStopsBeforeNextPipelineWindow(t *testing.T) {
|
|
total := queuePushBatchSize + queuePushBatchSize*queuePipelineBatchCount + 1
|
|
ids := make([]string, total)
|
|
for i := range ids {
|
|
ids[i] = fmt.Sprintf("video-%d", i)
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
client := &fakePublishClient{cancelAfterPipelineAt: 1, cancel: cancel}
|
|
|
|
err := publish(ctx, client, publishTestMeta(len(ids)), ids, "lock", "token")
|
|
if !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("publish() error = %v, want context.Canceled", err)
|
|
}
|
|
if client.pipelineCalls != 1 {
|
|
t.Fatalf("pipeline calls = %d, want 1", client.pipelineCalls)
|
|
}
|
|
if client.swapped || client.swapAttempts != 0 {
|
|
t.Fatalf("swap state = %v, attempts = %d, want no swap", client.swapped, client.swapAttempts)
|
|
}
|
|
if client.deleteCalls == 0 {
|
|
t.Fatal("cancelled building queue was not cleaned")
|
|
}
|
|
}
|
|
|
|
func TestPublishRejectsMetadataLengthMismatchBeforeRedis(t *testing.T) {
|
|
meta := publishTestMeta(2)
|
|
meta.Length = 1
|
|
|
|
err := publish(context.Background(), nil, meta, []string{"video-1", "video-2"}, "lock", "token")
|
|
if err == nil || !strings.Contains(err.Error(), "metadata length") {
|
|
t.Fatalf("publish() error = %v, want metadata length mismatch", err)
|
|
}
|
|
}
|
|
|
|
func publishTestMeta(length int) QueueMeta {
|
|
generatedAt := time.Date(2026, 7, 31, 2, 0, 0, 0, time.UTC)
|
|
return QueueMeta{
|
|
Version: VersionAt(generatedAt),
|
|
Length: length,
|
|
GeneratedAt: generatedAt,
|
|
StartOffset: 0,
|
|
}
|
|
}
|
|
|
|
func TestVersionValidUntilUsesCSTDayBoundary(t *testing.T) {
|
|
meta := QueueMeta{
|
|
Version: "20260731",
|
|
GeneratedAt: time.Date(2026, 7, 31, 15, 59, 59, 0, time.UTC),
|
|
}
|
|
want := time.Date(2026, 7, 31, 16, 0, 0, 0, time.UTC)
|
|
if got := versionValidUntil(meta); !got.Equal(want) {
|
|
t.Fatalf("versionValidUntil() = %s, want %s", got, want)
|
|
}
|
|
}
|
|
|
|
func TestAssembleConcurrentDeterministic(t *testing.T) {
|
|
now := time.Date(2026, 7, 24, 8, 0, 0, 0, time.UTC)
|
|
input := make([]vidmod.RecommendCandidate, 0, 100)
|
|
for i := 0; i < 100; i++ {
|
|
reviewAt := now.Add(-48 * time.Hour)
|
|
if i >= 90 {
|
|
reviewAt = now.Add(-time.Duration(i-90) * time.Hour)
|
|
}
|
|
input = append(input, candidate(i, int64(1000-i), reviewAt))
|
|
}
|
|
want := fmt.Sprint(Assemble(input, now))
|
|
var wg sync.WaitGroup
|
|
errs := make(chan string, 64)
|
|
for i := 0; i < 64; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
if got := fmt.Sprint(Assemble(input, now)); got != want {
|
|
errs <- got
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
close(errs)
|
|
if got, ok := <-errs; ok {
|
|
t.Fatalf("concurrent result differs: %s", got)
|
|
}
|
|
}
|
|
|
|
func assertUnique(t *testing.T, ids []string) {
|
|
t.Helper()
|
|
seen := map[string]bool{}
|
|
for _, id := range ids {
|
|
if seen[id] {
|
|
t.Fatalf("duplicate id %s", id)
|
|
}
|
|
seen[id] = true
|
|
}
|
|
}
|