package index
import (
"runtime"
"sync"
"sync/atomic"
"testing"
)
// A window asked for again is answered from memory until a fold lands:
// the activity query reads every touch in the window, and the page that
// shows it is the one a crawler swarm asks for.
func TestActivityIsRememberedUntilAFold(t *testing.T) {
ix := mergeDAG(t)
var runs atomic.Int32
ix.activityRun = func() { runs.Add(1) }
q := ActivityQuery{Since: "2026-06-01", Until: "2026-06-30"}
first, err := ix.Activity(q)
if err != nil {
t.Fatal(err)
}
again, err := ix.Activity(q)
if err != nil {
t.Fatal(err)
}
if runs.Load() != 1 || len(again.Rows) != len(first.Rows) {
t.Fatalf("FAIL: two asks ran the query %d times, want 1", runs.Load())
}
// Another window is its own answer.
if _, err := ix.Activity(ActivityQuery{Since: "2026-06-02", Until: "2026-06-30"}); err != nil {
t.Fatal(err)
}
if runs.Load() != 2 {
t.Fatalf("FAIL: a different window ran %d queries in all, want 2", runs.Load())
}
// A fold makes every remembered window stale.
ix.invalidateCounts()
if _, err := ix.Activity(q); err != nil {
t.Fatal(err)
}
if runs.Load() != 3 {
t.Fatalf("FAIL: after a fold the window ran %d queries in all, want 3", runs.Load())
}
t.Log("ok")
}
// Requests for one window that arrive while it is being computed wait
// for that computation: a burst costs one query, not one each.
func TestConcurrentActivityRequestsShareOneQuery(t *testing.T) {
ix := mergeDAG(t)
var runs atomic.Int32
release := make(chan struct{})
started := make(chan struct{}, 1)
ix.activityRun = func() {
if runs.Add(1) == 1 {
started <- struct{}{}
<-release
}
}
q := ActivityQuery{Since: "2026-06-01", Until: "2026-06-30"}
var wg sync.WaitGroup
results := make([]int, 8)
wg.Add(1)
go func() {
defer wg.Done()
a, err := ix.Activity(q)
if err == nil {
results[0] = len(a.Rows)
}
}()
<-started
for i := 1; i < len(results); i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
a, err := ix.Activity(q)
if err == nil {
results[i] = len(a.Rows)
}
}(i)
}
// Let the followers reach the wait before the leader finishes.
for ix.activityWaiters(q) < len(results)-1 {
runtimeYield()
}
close(release)
wg.Wait()
if runs.Load() != 1 {
t.Fatalf("FAIL: %d concurrent asks ran %d queries, want 1", len(results), runs.Load())
}
for i, n := range results {
if n != 3 {
t.Fatalf("FAIL: request %d got %d rows, want the 3 the leader read", i, n)
}
}
t.Log("ok")
}
// activityWaiters counts the requests waiting on q's computation.
func (ix *Index) activityWaiters(q ActivityQuery) int {
ix.activityMu.Lock()
defer ix.activityMu.Unlock()
if c, ok := ix.activityCalls[q]; ok {
return c.waiters
}
return 0
}
func runtimeYield() { runtime.Gosched() }