243 lines
7.2 KiB
Go
243 lines
7.2 KiB
Go
package store
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"sync"
|
||
"sync/atomic"
|
||
"testing"
|
||
"time"
|
||
|
||
"telesrv/internal/domain"
|
||
)
|
||
|
||
type blockingReadModelVersionStore struct {
|
||
calls atomic.Int32
|
||
once sync.Once
|
||
started chan struct{}
|
||
release chan struct{}
|
||
hashes map[ReadModelKey]int64
|
||
mu sync.Mutex
|
||
loaded map[ReadModelKey]int
|
||
}
|
||
|
||
func (s *blockingReadModelVersionStore) ReadModelHash(ctx context.Context, model string, ownerUserID int64, peerType domain.PeerType, peerID int64) (int64, bool, error) {
|
||
rows, err := s.ReadModelHashes(ctx, []ReadModelKey{{Model: model, OwnerUserID: ownerUserID, PeerType: peerType, PeerID: peerID}})
|
||
if err != nil {
|
||
return 0, false, err
|
||
}
|
||
key := ReadModelKey{Model: model, OwnerUserID: ownerUserID, PeerType: peerType, PeerID: peerID}
|
||
hash := rows[key]
|
||
return hash, hash != 0, nil
|
||
}
|
||
|
||
func (s *blockingReadModelVersionStore) ReadModelHashes(ctx context.Context, keys []ReadModelKey) (map[ReadModelKey]int64, error) {
|
||
s.calls.Add(1)
|
||
s.once.Do(func() { close(s.started) })
|
||
select {
|
||
case <-s.release:
|
||
case <-ctx.Done():
|
||
return nil, ctx.Err()
|
||
}
|
||
out := make(map[ReadModelKey]int64, len(keys))
|
||
s.mu.Lock()
|
||
if s.loaded == nil {
|
||
s.loaded = make(map[ReadModelKey]int)
|
||
}
|
||
for _, key := range keys {
|
||
s.loaded[key]++
|
||
if hash := s.hashes[key]; hash != 0 {
|
||
out[key] = hash
|
||
}
|
||
}
|
||
s.mu.Unlock()
|
||
return out, nil
|
||
}
|
||
|
||
func TestCachedReadModelVersionStoreSingleflightsBatchMiss(t *testing.T) {
|
||
ctx := context.Background()
|
||
keys := []ReadModelKey{
|
||
{Model: "channel_base", OwnerUserID: 0, PeerType: domain.PeerTypeChannel, PeerID: 10},
|
||
{Model: "channel_member", OwnerUserID: 100, PeerType: domain.PeerTypeChannel, PeerID: 10},
|
||
}
|
||
base := &blockingReadModelVersionStore{
|
||
started: make(chan struct{}),
|
||
release: make(chan struct{}),
|
||
hashes: map[ReadModelKey]int64{
|
||
keys[0]: 11,
|
||
keys[1]: 22,
|
||
},
|
||
}
|
||
cache := NewCachedReadModelVersionStore(base, time.Hour, 100)
|
||
|
||
const goroutines = 16
|
||
errs := make(chan error, goroutines)
|
||
var wg sync.WaitGroup
|
||
wg.Add(goroutines)
|
||
for i := 0; i < goroutines; i++ {
|
||
go func() {
|
||
defer wg.Done()
|
||
rows, err := cache.ReadModelHashes(ctx, keys)
|
||
if err != nil {
|
||
errs <- err
|
||
return
|
||
}
|
||
if rows[keys[0]] != 11 || rows[keys[1]] != 22 {
|
||
errs <- fmt.Errorf("hashes = %+v, want 11/22", rows)
|
||
}
|
||
}()
|
||
}
|
||
<-base.started
|
||
time.Sleep(20 * time.Millisecond)
|
||
close(base.release)
|
||
wg.Wait()
|
||
close(errs)
|
||
for err := range errs {
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
if got := base.calls.Load(); got != 1 {
|
||
t.Fatalf("base ReadModelHashes calls = %d, want 1", got)
|
||
}
|
||
if _, err := cache.ReadModelHashes(ctx, keys); err != nil {
|
||
t.Fatalf("cached ReadModelHashes: %v", err)
|
||
}
|
||
if got := base.calls.Load(); got != 1 {
|
||
t.Fatalf("cache hit called base again: calls=%d", got)
|
||
}
|
||
}
|
||
|
||
func TestCachedReadModelVersionStoreWaitsOverlappingBatchKeys(t *testing.T) {
|
||
ctx := context.Background()
|
||
a := ReadModelKey{Model: "channel_base", OwnerUserID: 0, PeerType: domain.PeerTypeChannel, PeerID: 10}
|
||
b := ReadModelKey{Model: "channel_member", OwnerUserID: 100, PeerType: domain.PeerTypeChannel, PeerID: 10}
|
||
c := ReadModelKey{Model: "dialog_light", OwnerUserID: 100, PeerType: domain.PeerTypeChannel, PeerID: 10}
|
||
base := &blockingReadModelVersionStore{
|
||
started: make(chan struct{}),
|
||
release: make(chan struct{}),
|
||
hashes: map[ReadModelKey]int64{
|
||
a: 11,
|
||
b: 22,
|
||
c: 33,
|
||
},
|
||
}
|
||
cache := NewCachedReadModelVersionStore(base, time.Hour, 100)
|
||
|
||
firstErr := make(chan error, 1)
|
||
go func() {
|
||
rows, err := cache.ReadModelHashes(ctx, []ReadModelKey{a, b})
|
||
if err == nil && (rows[a] != 11 || rows[b] != 22) {
|
||
err = fmt.Errorf("first hashes = %+v, want A/B", rows)
|
||
}
|
||
firstErr <- err
|
||
}()
|
||
<-base.started
|
||
|
||
secondErr := make(chan error, 1)
|
||
go func() {
|
||
rows, err := cache.ReadModelHashes(ctx, []ReadModelKey{b, c})
|
||
if err == nil && (rows[b] != 22 || rows[c] != 33) {
|
||
err = fmt.Errorf("second hashes = %+v, want B/C", rows)
|
||
}
|
||
secondErr <- err
|
||
}()
|
||
time.Sleep(20 * time.Millisecond)
|
||
close(base.release)
|
||
|
||
if err := <-firstErr; err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := <-secondErr; err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
base.mu.Lock()
|
||
loadedB := base.loaded[b]
|
||
loadedC := base.loaded[c]
|
||
base.mu.Unlock()
|
||
if loadedB != 1 {
|
||
t.Fatalf("overlap key B loaded %d times, want 1", loadedB)
|
||
}
|
||
if loadedC != 1 {
|
||
t.Fatalf("owned key C loaded %d times, want 1", loadedC)
|
||
}
|
||
}
|
||
|
||
// TestCachedReadModelVersionStoreEpochGuardRejectsStaleWriteback 证明 epoch 守卫堵住了
|
||
// lost-update：一次锁外 DB load 期间到达的 NOTIFY(写入新 hash)不得被 load 返回的旧 hash 覆盖。
|
||
// 这是 4 个 hash-only 消费缓存(private/channel media counts、participants、active ids)正确性的根。
|
||
func TestCachedReadModelVersionStoreEpochGuardRejectsStaleWriteback(t *testing.T) {
|
||
ctx := context.Background()
|
||
key := ReadModelKey{Model: "private_media_counts", OwnerUserID: 100, PeerType: domain.PeerTypeUser, PeerID: 100}
|
||
base := &blockingReadModelVersionStore{
|
||
started: make(chan struct{}),
|
||
release: make(chan struct{}),
|
||
hashes: map[ReadModelKey]int64{key: 11}, // DB 此刻仍是旧 hash 11
|
||
}
|
||
cache := NewCachedReadModelVersionStore(base, time.Hour, 100)
|
||
|
||
resultCh := make(chan map[ReadModelKey]int64, 1)
|
||
errCh := make(chan error, 1)
|
||
go func() {
|
||
rows, err := cache.ReadModelHashes(ctx, []ReadModelKey{key})
|
||
if err != nil {
|
||
errCh <- err
|
||
return
|
||
}
|
||
resultCh <- rows
|
||
}()
|
||
|
||
// 等到 load 已 claim inflight 并阻塞在 base.ReadModelHashes 内。
|
||
<-base.started
|
||
// 模拟 NOTIFY 在 load 期间送来权威新 hash 99（写入并自增 epoch）。
|
||
cache.UpdateReadModelHash(key, 99)
|
||
// 放行 load：它会返回旧 hash 11；epoch 守卫必须拒绝用 11 覆盖 99。
|
||
close(base.release)
|
||
|
||
select {
|
||
case err := <-errCh:
|
||
t.Fatalf("ReadModelHashes: %v", err)
|
||
case rows := <-resultCh:
|
||
if rows[key] != 99 {
|
||
t.Fatalf("in-flight reader got hash %d, want fresh 99 (stale 11 must not win)", rows[key])
|
||
}
|
||
case <-time.After(2 * time.Second):
|
||
t.Fatal("timeout waiting for in-flight reader")
|
||
}
|
||
|
||
// 后续读仍须是 99，且不再打 base —— 证明缓存没被旧值污染、也没被错误驱逐。
|
||
callsBefore := base.calls.Load()
|
||
rows, err := cache.ReadModelHashes(ctx, []ReadModelKey{key})
|
||
if err != nil {
|
||
t.Fatalf("second ReadModelHashes: %v", err)
|
||
}
|
||
if rows[key] != 99 {
|
||
t.Fatalf("cached hash = %d, want 99 (cache poisoned by stale 11)", rows[key])
|
||
}
|
||
if got := base.calls.Load(); got != callsBefore {
|
||
t.Fatalf("second read hit base (calls %d -> %d): cache was poisoned or evicted", callsBefore, got)
|
||
}
|
||
}
|
||
|
||
func TestCachedReadModelVersionStoreUpdateReadModelHashWarmsCache(t *testing.T) {
|
||
ctx := context.Background()
|
||
key := ReadModelKey{Model: "channel_base", PeerType: domain.PeerTypeChannel, PeerID: 10}
|
||
base := &blockingReadModelVersionStore{
|
||
started: make(chan struct{}),
|
||
release: make(chan struct{}),
|
||
hashes: map[ReadModelKey]int64{key: 11},
|
||
}
|
||
cache := NewCachedReadModelVersionStore(base, time.Hour, 100)
|
||
|
||
cache.UpdateReadModelHash(key, 99)
|
||
rows, err := cache.ReadModelHashes(ctx, []ReadModelKey{key})
|
||
if err != nil {
|
||
t.Fatalf("ReadModelHashes: %v", err)
|
||
}
|
||
if rows[key] != 99 {
|
||
t.Fatalf("hash = %d, want notify-provided 99", rows[key])
|
||
}
|
||
if got := base.calls.Load(); got != 0 {
|
||
t.Fatalf("base calls = %d, want cache warmed by notify", got)
|
||
}
|
||
}
|