Something went wrong. Try again.
An ATProto Powered Reputation Network
Something went wrong. Try again.
7.0 kB · 200 lines
Go
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201package appview
import ( "context" "errors" "reflect" "sync" "sync/atomic" "testing" "time"
"provenance/internal/domain" "provenance/internal/storage" "provenance/internal/trust" "provenance/internal/trust/algorithms")
type countedTrust struct { algorithm trust.Algorithm calls atomic.Int64 err error}
func (a *countedTrust) Describe() trust.Descriptor { return a.algorithm.Describe() }func (a *countedTrust) Calculate(ctx context.Context, input trust.Input) (trust.Result, error) { a.calls.Add(1) if a.err != nil { return trust.Result{}, a.err } return a.algorithm.Calculate(ctx, input)}
type graphCounter struct { storage.Store reads atomic.Int64}
func (s *graphCounter) TrustGraph(ctx context.Context, tag string, mode domain.Mode, limit int) ([]domain.Vouch, int64, error) { s.reads.Add(1) return s.Store.TrustGraph(ctx, tag, mode, limit)}
func TestCachedTrustQueries(t *testing.T) { algorithm := &countedTrust{algorithm: algorithms.ShortestPath{}} s, _ := algorithmService(t, algorithm) store := &graphCounter{Store: s.store} s.store = store now := time.UnixMicro(time.Now().UnixMicro()) s.now = func() time.Time { return now } ctx := context.Background() first, err := s.Reputation(ctx, alice, bob, "", domain.Default) if err != nil { t.Fatal(err) } again, err := s.Reputation(ctx, "alice.example.com", "bob.example.com", "", domain.Default) if err != nil || !reflect.DeepEqual(first, again) || algorithm.calls.Load() != 1 || store.reads.Load() != 1 { t.Fatalf("cache miss: %+v %v; calls %d reads %d", again, err, algorithm.calls.Load(), store.reads.Load()) } if err := s.SetThreshold(ctx, alice, -2); err != nil { t.Fatal(err) } if _, err := s.Evaluate(ctx, alice, bob, "", domain.Default); err != nil { t.Fatal(err) } if algorithm.calls.Load() != 1 { t.Fatal("threshold change invalidated a trust calculation") } // A newly added alternative path remains invisible until the cached result expires. v := domain.Vouch{Author: alice, Subject: bob, Score: 2} if err := store.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: alice, RecordKey: domain.RecordKey(bob, ""), TimeUS: time.Now().Add(time.Minute).UnixMicro(), Vouch: &v}); err != nil { t.Fatal(err) } again, err = s.Reputation(ctx, alice, bob, "", domain.Default) if err != nil || !reflect.DeepEqual(first, again) { t.Fatalf("new path invalidated existing cache: %+v %v", again, err) } now = now.Add(s.cacheTTL) updated, err := s.Reputation(ctx, alice, bob, "", domain.Default) if err != nil || updated.Score == nil || *updated.Score != 2 || algorithm.calls.Load() != 2 { t.Fatalf("expired result: %+v %v", updated, err) } if _, err := s.Reputation(ctx, alice, alice, "", domain.Default); err != nil { t.Fatal(err) } if _, err := s.Reputation(ctx, alice, alice, "", domain.Default); err != nil { t.Fatal(err) } if algorithm.calls.Load() != 3 { t.Fatal("self result was not cached") } canceled, cancel := context.WithCancel(ctx) cancel() if _, err := s.Reputation(canceled, alice, bob, "", domain.Default); !errors.Is(err, context.Canceled) { t.Fatalf("cached cancellation: %v", err) }}
func TestCacheKeyIsolation(t *testing.T) { algorithm := &countedTrust{algorithm: algorithms.ShortestPath{}} s, _ := algorithmService(t, algorithm) ctx := context.Background() for _, options := range []trust.Accounts{ {algorithms.IgnoredParameter: {carol, dan}}, {algorithms.IgnoredParameter: {dan, carol}}, } { if _, err := s.Reputation(ctx, alice, bob, "", domain.Default, options); err != nil { t.Fatal(err) } } if algorithm.calls.Load() != 1 { t.Fatal("account order changed the cache key") } if _, err := s.Reputation(ctx, alice, bob, "", domain.Default); err != nil { t.Fatal(err) } if _, err := s.Reputation(ctx, alice, bob, "work", domain.Fallback); err != nil { t.Fatal(err) } if _, err := s.Reputation(ctx, alice, bob, "work", domain.Tagged); err != nil { t.Fatal(err) } if algorithm.calls.Load() != 4 { t.Fatal("options or topic mode shared an incorrect cache entry") } other := NewWithAlgorithm(s.store, s.repo, s.limits, algorithms.Optimistic{}) got, err := other.Reputation(ctx, alice, bob, "", domain.Default) if err != nil || got.Score == nil || *got.Score != 1.8 { t.Fatalf("algorithms shared a cache entry: %+v %v", got, err) } limits := s.limits limits.Penalty = 0 noPenalty := NewWithAlgorithm(s.store, s.repo, limits, algorithms.Optimistic{}) got, err = noPenalty.Reputation(ctx, alice, bob, "", domain.Default) if err != nil || got.Score == nil || *got.Score != 2 { t.Fatalf("limits shared a cache entry: %+v %v", got, err) } for _, pair := range [][2]string{{carol, bob}, {alice, dan}} { got, err = other.Reputation(ctx, pair[0], pair[1], "", domain.Default) if err != nil || got.Score == nil || *got.Score != 2 { t.Fatalf("identities shared a cache entry: %+v %v", got, err) } }}
func TestFailedTrustNotCached(t *testing.T) { failure := errors.New("calculation failed") algorithm := &countedTrust{algorithm: algorithms.Optimistic{}, err: failure} s, _ := algorithmService(t, algorithm) for range 2 { if _, err := s.Reputation(context.Background(), alice, bob, "", domain.Default); !errors.Is(err, failure) { t.Fatal(err) } } if algorithm.calls.Load() != 2 { t.Fatal("cached a calculation error") }}
func TestCachePublicationRace(t *testing.T) { s, _ := algorithmService(t, algorithms.Optimistic{}) calls := 0 s.algorithm = testAlgorithm{calculate: func(ctx context.Context, edges []domain.Vouch, source, target string, limits trust.Limits) (trust.Result, error) { calls++ if calls == 1 { v := domain.Vouch{Author: alice, Subject: bob, Score: 2} if err := s.store.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: alice, RecordKey: domain.RecordKey(bob, ""), TimeUS: time.Now().Add(time.Minute).UnixMicro(), Vouch: &v}); err != nil { return trust.Result{}, err } } return (algorithms.Optimistic{}).Calculate(ctx, trust.Input{Edges: edges, Source: source, Target: target, Limits: limits}) }} for range 2 { got, err := s.Reputation(context.Background(), alice, bob, "", domain.Default) if err != nil || got.Score == nil || *got.Score != 2 { t.Fatalf("stale publication: %+v %v", got, err) } } if calls != 2 { t.Fatalf("calculations = %d, want 2", calls) }}
func TestConcurrentTrustReader(t *testing.T) { algorithm := &countedTrust{algorithm: algorithms.Optimistic{}} s, _ := algorithmService(t, algorithm) reader := &trustReader{service: s, input: trust.Input{Source: alice, Target: bob, Limits: s.limits}, mode: domain.Default} var workers sync.WaitGroup for range 8 { workers.Go(func() { got, err := reader.Score(context.Background(), bob) if err != nil || got.Score == nil || *got.Score != 1.8 { t.Errorf("concurrent trust: %+v %v", got, err) } }) } workers.Wait() if algorithm.calls.Load() != 1 { t.Fatal("concurrent author queries did not share their cached result") }}