package main
import (
"context"
"testing"
"time"
"craftdesign.group/quarry/internal/errlog"
"craftdesign.group/quarry/internal/ingest"
"craftdesign.group/quarry/internal/work"
)
// journalled counts the ingest journal's reports for one store.
func journalled(repo string) int {
n := 0
for _, e := range errlog.Default.Recent(500) {
if e.Surface == "ingest" && e.Repo == repo {
n += e.Count
}
}
return n
}
// While boot rebuilds held the ingest lane, every other store's poll
// round was shed every two seconds, and each shed was journalled as "a
// store could not be ingested": 613 entries in minutes, burying real
// errors. A shed is back-pressure: counted, retried with backoff, and
// journalled once, only when the store has stayed unindexed past the
// deadline.
func TestAShedIngestRoundBacksOffAndIsJournalledOncePastTheDeadline(t *testing.T) {
const repo = "shedstore"
lane := work.NewLane("ingest", 1, 0)
rg := ®istrar{ingestLane: lane, interval: 2 * time.Second}
hold, held := make(chan struct{}), make(chan struct{})
go lane.Do(context.Background(), func() error { close(held); <-hold; return nil })
<-held
t.Cleanup(func() {
select {
case <-hold:
default:
close(hold)
}
})
// The journal is process-wide and outlives a -count run.
before := journalled(repo)
var shed shedBackoff
round := func(now time.Time) bool {
ran, err := rg.ingestRound(context.Background(), now, repo, &shed, func(ingest.Aside) error { return nil })
if err != nil {
t.Fatalf("a shed round returned %v", err)
}
return ran
}
t0 := time.Now()
if round(t0) {
t.Fatal("a round ran in a full lane")
}
if got := lane.Stats().Shed; got != 1 {
t.Errorf("FAIL: the shed counter is %d, want 1", got)
}
if round(t0.Add(time.Second)); lane.Stats().Shed != 1 {
t.Errorf("FAIL: a round was retried inside its backoff: shed %d", lane.Stats().Shed)
} else {
t.Log("ok: a shed round is counted and not retried inside its backoff")
}
// Rounds keep being shed, each after its backoff, until well past
// the deadline.
sheds := 1
for now := t0; now.Sub(t0) < 3*ingestShedDeadline; now = now.Add(time.Second) {
if !round(now) && lane.Stats().Shed > int64(sheds) {
sheds = int(lane.Stats().Shed)
}
}
if sheds >= int(3*ingestShedDeadline/(2*time.Second)) {
t.Errorf("FAIL: %d attempts in %s: the retries do not back off", sheds, 3*ingestShedDeadline)
}
if got := journalled(repo) - before; got != 1 {
t.Errorf("FAIL: the journal holds %d reports for a store shed past the deadline, want 1", got)
} else {
t.Logf("ok: %d sheds over %s journalled once", sheds, 3*ingestShedDeadline)
}
// A round that gets in ends the episode.
close(hold)
for deadline := time.Now().Add(5 * time.Second); lane.Stats().InFlight > 0; time.Sleep(5 * time.Millisecond) {
if time.Now().After(deadline) {
t.Fatal("the blocking unit never left the lane")
}
}
later := t0.Add(time.Hour)
for i := 0; !round(later); i++ {
later = later.Add(time.Minute)
if i > 10 {
t.Fatal("a round never got into an empty lane")
}
}
if !shed.since.IsZero() {
t.Errorf("FAIL: a round that ran did not end the shed episode: %+v", shed)
} else {
t.Log("ok: a round that runs ends the episode")
}
}