//go:build torture

package spindle

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/bluesky-social/indigo/atproto/atcrypto"
	"github.com/bluesky-social/indigo/atproto/auth"
	"github.com/bluesky-social/indigo/atproto/syntax"

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/spindle/models"
)

func (w *tortureWorld) abuseRepos(ks []*tortureKnot, tag string, perKnot, commits int, workflows map[string]string) []*tortureRepo {
	return w.addRepos(len(ks)*perKnot, func(i int) (*tortureKnot, syntax.DID) {
		idx := i % len(ks)
		return ks[idx], syntax.DID(fmt.Sprintf("did:plc:%s%d", tag, idx))
	}, commits, workflows)
}

func abuseQueues(repos []*tortureRepo, from, to int) map[*tortureKnot][]pushDelivery {
	queues := map[*tortureKnot][]pushDelivery{}
	for i := from; i < to; i++ {
		for _, repo := range repos {
			queues[repo.knot] = append(queues[repo.knot], pushRange(repo, i, i+1)...)
		}
	}
	return queues
}

func abuseRatio(n, of int64) float64 {
	if of == 0 {
		return 0
	}
	return float64(n) / float64(of)
}

func (w *tortureWorld) abuseHammer(ctx context.Context, conc int, token func() string, body []byte, st *deliveryStats) {
	client := &http.Client{Timeout: 30 * time.Second, Transport: &http.Transport{MaxIdleConnsPerHost: conc}}
	defer client.CloseIdleConnections()
	var wg sync.WaitGroup
	for range conc {
		wg.Go(func() {
			for ctx.Err() == nil {
				st.attempts.Add(1)
				reply := w.post(ctx, client, token(), body)
				if ctx.Err() != nil {
					return
				}
				st.saw(reply.status)
				st.first.add(reply.elapsed)
			}
		})
	}
	wg.Wait()
}

type abuseHook struct {
	srv  *httptest.Server
	got  atomic.Int64
	stop chan struct{}
	once sync.Once

	mu        sync.Mutex
	firstSeen map[string]time.Time
}

func newAbuseHook(r *tortureRun, hang time.Duration) *abuseHook {
	h := &abuseHook{stop: make(chan struct{}), firstSeen: map[string]time.Time{}}
	h.srv = httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
		raw, _ := io.ReadAll(io.LimitReader(req.Body, 1<<20))
		if hang > 0 {
			select {
			case <-time.After(hang):
			case <-h.stop:
			case <-req.Context().Done():
			}
		}
		var p struct {
			After      string `json:"after"`
			Repository struct {
				FullName string `json:"full_name"`
			} `json:"repository"`
		}
		_ = json.Unmarshal(raw, &p)
		h.got.Add(1)
		key := p.After + "|" + p.Repository.FullName
		h.mu.Lock()
		if _, ok := h.firstSeen[key]; !ok {
			h.firstSeen[key] = time.Now()
		}
		h.mu.Unlock()
		rw.WriteHeader(http.StatusOK)
	}))
	// cleanups run in reverse, and srv.Close waits on hung handlers unless release runs first
	r.t.Cleanup(h.srv.Close)
	r.t.Cleanup(h.release)
	return h
}

func (h *abuseHook) release() { h.once.Do(func() { close(h.stop) }) }

func (h *abuseHook) seenAt(key string) (time.Time, bool) {
	h.mu.Lock()
	defer h.mu.Unlock()
	at, ok := h.firstSeen[key]
	return at, ok
}

func abuseHookKey(d pushDelivery) string {
	return d.newSha + "|" + d.repo.owner.String() + "/" + strings.ToLower(d.repo.name)
}

// straight to the db, skips the per repo and per owner quotas
func (w *tortureWorld) abuseAddHook(repo *tortureRepo, url string) {
	hook := &models.Webhook{RepoDid: repo.did, Url: url, Active: true, Events: []string{string(models.WebhookEventPush)}}
	created, err := w.s.db.AddWebhook(context.Background(), hook, 0, 0)
	if err != nil || !created {
		w.r.t.Fatalf("add webhook: created=%v err=%v", created, err)
	}
}

func (w *tortureWorld) abuseNotifyInflight() float64 {
	return w.fromSpindle(func(s *Spindle) float64 { return float64(s.notifySlots.inflight()) })
}

func (w *tortureWorld) abuseUnknownInflight() float64 {
	return w.fromSpindle(func(s *Spindle) float64 {
		s.notifyAdmission.mu.Lock()
		defer s.notifyAdmission.mu.Unlock()
		return float64(s.notifyAdmission.total)
	})
}

func (w *tortureWorld) abuseHeld(host string) float64 {
	return w.fromSpindle(func(s *Spindle) float64 {
		s.notifySlots.mu.Lock()
		defer s.notifySlots.mu.Unlock()
		return float64(s.notifySlots.held[host])
	})
}

func (w *tortureWorld) abuseSybil(ctx context.Context, k *tortureKnot, repos []*tortureRepo, next *atomic.Int64, st *deliveryStats) {
	client := &http.Client{Timeout: time.Minute}
	sha := strings.Repeat("a", 40)
	for ctx.Err() == nil {
		repo := repos[int(next.Add(1))%len(repos)]
		st.attempts.Add(1)
		reply := w.post(ctx, client, w.token(k), w.notifyBody(repo, "refs/heads/main", zeroSha, sha))
		if ctx.Err() != nil {
			return
		}
		st.saw(reply.status)
		if reply.status == http.StatusServiceUnavailable {
			sleepCtx(ctx, 200*time.Millisecond)
		}
	}
}

func init() {
	// shorter than the knot client's 30s timeout on purpose, which keeps slots turning over faster
	registerTorture("notify-sybil", "hostile hosts notifying for repos nobody serves while real knots flush outboxes", func(r *tortureRun) {
		hosts := r.load("hostile_hosts", 5)
		conc := 6
		hold := r.duration("hostile_hold", 6*time.Second)
		perHost := 40
		legitKnots := 4
		perKnot := 2
		pushes := int(r.param("pushes", 12))
		budget := r.duration("legit_budget", 75*time.Second)

		w := newTortureWorld(r)
		w.maxJobs = 8
		r.gauge("notify_inflight", w.abuseNotifyInflight)
		r.gauge("notify_unknown_inflight", w.abuseUnknownInflight)
		legit := w.addKnots(legitKnots)
		rs := w.abuseRepos(legit, "sybillegit", perKnot, pushes, map[string]string{"ci.yaml": tortureWorkflow("sleep 20ms")})
		hostile := w.addKnots(hosts)
		hostileRepos := map[*tortureKnot][]*tortureRepo{}
		for i, k := range hostile {
			for j := range perHost {
				owner := syntax.DID(fmt.Sprintf("did:plc:sybilowner%d", i*100000+j))
				hostileRepos[k] = append(hostileRepos[k], w.addRepo(k, owner, 0, nil))
			}
		}
		w.boot()
		w.seed(rs, 3*time.Minute)

		half := pushes / 2
		var base deliveryStats
		start := time.Now()
		w.outboxes(context.Background(), abuseQueues(rs, 0, half), &base)
		baseline := time.Since(start)
		r.metric("baseline_flush_s", baseline.Seconds())
		r.check("baseline delivers everything", base.accepted.len() == len(rs)*half, "%d of %d accepted before any attack", base.accepted.len(), len(rs)*half)

		atkCtx, stopAtk := context.WithCancel(context.Background())
		defer stopAtk()
		var hst deliveryStats
		var atk sync.WaitGroup
		for _, k := range hostile {
			k.delay.Store(int64(hold))
			next := &atomic.Int64{}
			for range conc {
				atk.Go(func() { w.abuseSybil(atkCtx, k, hostileRepos[k], next, &hst) })
			}
		}
		want := float64(min(notifyMaxUnknown, hosts*notifyMaxUnknownPerKnot))
		saturated := r.waitFor(3*hold+5*time.Second, func() bool { return w.abuseUnknownInflight() >= want })
		r.metric("unknown_pool_saturated", map[bool]float64{true: 1, false: 0}[saturated])
		r.check("hostile knots fill the pool for unknown repos (setup)", saturated, "%.0f of %.0f unknown-repo seats held", w.abuseUnknownInflight(), want)

		legitCtx, cancel := context.WithTimeout(context.Background(), budget)
		var st deliveryStats
		start = time.Now()
		w.outboxes(legitCtx, abuseQueues(rs, half, pushes), &st)
		attack := time.Since(start)
		cancel()
		stopAtk()
		atk.Wait()

		st.report(r, "legit_")
		hst.report(r, "hostile_")
		r.metric("attack_flush_s", attack.Seconds())
		r.metric("legit_busy_ratio", abuseRatio(int64(st.seen("http_503")), st.attempts.Load()))
		want2 := len(rs) * (pushes - half)
		maxWait := st.accepted.quantile(1)
		if st.accepted.len() < want2 {
			// pushes that never landed waited the whole attack, or they'd look instant
			maxWait = max(maxWait, attack)
		}
		r.metric("legit_max_wait_s", maxWait.Seconds())
		r.note("hostile hold is %v per call and ensureRepo makes three calls to the hostile knot, production clients hold up to 30s each. the notify slots stay free of them, they only hold the unknown-repo seats", hold)

		r.check("legit pushes all delivered under sybil load (N2)", st.accepted.len() == want2, "%d of %d legit pushes accepted within %v", st.accepted.len(), want2, budget)
		r.check("legit knots mostly not told busy (N2)", abuseRatio(int64(st.seen("http_503")), st.attempts.Load()) <= 0.1, "%d of %d legit attempts got 503, want at most 10%%", st.seen("http_503"), st.attempts.Load())
		r.check("legit flush time not blown up (N2)", attack <= 3*baseline+10*time.Second, "flush took %v, baseline %v, want <= 3x baseline + 10s", attack.Round(time.Millisecond), baseline.Round(time.Millisecond))
		r.check("no legit push waits more than 20s (N2)", maxWait <= 20*time.Second, "longest wait %v, want <= 20s", maxWait.Round(time.Millisecond))
	})

	registerTorture("notify-slot-share", "a knot with served repos and a hanging fetch takes the notify slots, then real knots arrive", func(r *tortureRun) {
		pool := int(r.param("slots", 16))
		hold := r.duration("fetch_hold", 5*time.Second)
		budget := r.duration("legit_budget", 90*time.Second)
		const commits = 6
		workflows := map[string]string{"ci.yaml": tortureWorkflow("sleep 20ms")}

		w := newTortureWorld(r)
		w.maxJobs = 8
		w.notifySlots = pool
		r.gauge("notify_inflight", w.abuseNotifyInflight)
		legit := w.addKnots(3)
		rs := w.abuseRepos(legit, "sharelegit", 2, commits, workflows)
		hostile := w.addKnots(1)
		hrs := w.abuseRepos(hostile, "sharehostile", 12, commits, workflows)
		w.boot()
		w.seed(append(slices.Clone(rs), hrs...), 3*time.Minute)
		hk := hostile[0]
		hk.gitDelay.Store(hold.Nanoseconds())

		atkCtx, stopAtk := context.WithCancel(context.Background())
		defer stopAtk()
		ds := pushRanges(hrs, 0, commits)
		var hst deliveryStats
		var atk sync.WaitGroup
		atk.Go(func() { w.flood(atkCtx, ds, 4*pool, 1000, &hst) })
		shared := pool - poolReserve(pool)
		reached := r.waitFor(30*time.Second, func() bool { return w.abuseHeld(hk.host()) >= float64(shared) })
		r.check("an idle pool lets one knot use all of it but the reserve", reached, "the hostile knot held %.0f of %d slots, want the %d that aren't the reserve while nobody else waits", w.abuseHeld(hk.host()), pool, shared)
		first := w.notifyWith(context.Background(), http.DefaultClient, pushRange(rs[0], 0, 1)[0])
		r.metric("first_legit_notify_ms", float64(first.elapsed.Milliseconds()))
		r.check("a real knot gets the reserve at once while a hanging knot holds the rest", first.status == http.StatusOK && first.elapsed <= 2*time.Second,
			"the first legit push answered %d after %v, want 200 within 2s", first.status, first.elapsed.Round(time.Millisecond))
		r.check("a hanging knot never holds the reserve", w.abuseHeld(hk.host()) <= float64(shared), "the hostile knot held %.0f of %d slots, want at most %d", w.abuseHeld(hk.host()), pool, shared)

		// peak only starts once the first hostile calls are done holding their slots
		var peak atomic.Int64
		stopSample := make(chan struct{})
		sampled := make(chan struct{})
		go func() {
			defer close(sampled)
			select {
			case <-time.After(hold + 2*time.Second):
			case <-stopSample:
				return
			}
			for {
				select {
				case <-stopSample:
					return
				case <-time.After(10 * time.Millisecond):
				}
				if held := int64(w.abuseHeld(hk.host())); held > peak.Load() {
					peak.Store(held)
				}
			}
		}()

		legitCtx, cancel := context.WithTimeout(context.Background(), budget)
		var st deliveryStats
		start := time.Now()
		w.outboxes(legitCtx, abuseQueues(rs, 0, commits), &st)
		took := time.Since(start)
		cancel()
		close(stopSample)
		<-sampled
		stopAtk()
		atk.Wait()

		st.report(r, "legit_")
		hst.report(r, "hostile_")
		r.metric("legit_flush_s", took.Seconds())
		r.metric("hostile_peak_held_after_hold", float64(peak.Load()))
		want := len(rs) * commits
		r.check("real knots get all their pushes delivered while a knot hangs on the slots", st.accepted.len() == want, "%d of %d legit pushes accepted within %v", st.accepted.len(), want, budget)
		share := int64(max(2, pool/2))
		r.check("a hanging knot is cut back to its share once others wait", peak.Load() <= share, "the hostile knot held up to %d of %d slots after the real knots were turned away, share with one other waiting is %d", peak.Load(), pool, share)
	})

	registerTorture("notify-bad-tokens", "a flood of forged tokens naming a real knot against that knot's own notifies", func(r *tortureRun) {
		conc := 16
		resolveDelay := 20 * time.Millisecond
		window := 4 * time.Second

		w := newTortureWorld(r)
		ks := w.addKnots(1)
		rs := w.abuseRepos(ks, "badtok", 2, 2, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")})
		w.boot()
		w.seed(rs, 3*time.Minute)
		k, repo := ks[0], rs[0]
		body := w.notifyBody(repo, "refs/heads/main", zeroSha, repo.shas[0])
		if reply := w.post(context.Background(), http.DefaultClient, w.token(k), body); reply.status != http.StatusOK {
			r.t.Fatalf("first legit notify answered %d: %s", reply.status, reply.body)
		}

		bad, err := atcrypto.GeneratePrivateKeyP256()
		if err != nil {
			r.t.Fatal(err)
		}
		lxm := syntax.NSID(org_tangled.EventNotifyEventNSID)
		sign := func(aud string, ttl time.Duration, key atcrypto.PrivateKey) func() string {
			return func() string {
				token, err := auth.SignServiceAuth(k.did, aud, ttl, &lxm, key)
				if err != nil {
					panic(err)
				}
				return token
			}
		}
		forged := sign(w.spindleDid(), time.Minute, bad)

		w.ids.delay.Store(int64(resolveDelay))
		probe := func(ctx context.Context, lat *latencies, notOK *atomic.Int64) {
			client := &http.Client{Timeout: 10 * time.Second}
			defer client.CloseIdleConnections()
			tick := time.NewTicker(50 * time.Millisecond)
			defer tick.Stop()
			for {
				select {
				case <-ctx.Done():
					return
				case <-tick.C:
				}
				reply := w.post(ctx, client, w.token(k), body)
				if ctx.Err() != nil {
					return
				}
				lat.add(reply.elapsed)
				if reply.status != http.StatusOK {
					notOK.Add(1)
				}
			}
		}
		var baseLat, floodLat latencies
		var baseBad, floodBad atomic.Int64
		baseCtx, cancelBase := context.WithTimeout(context.Background(), window)
		probe(baseCtx, &baseLat, &baseBad)
		cancelBase()

		floodCtx, cancelFlood := context.WithTimeout(context.Background(), window)
		var flood deliveryStats
		var wg sync.WaitGroup
		wg.Go(func() { w.abuseHammer(floodCtx, conc, forged, body, &flood) })
		wg.Go(func() { probe(floodCtx, &floodLat, &floodBad) })
		wg.Wait()
		cancelFlood()
		w.ids.delay.Store(0)

		r.rate("forged_flood_requests", int(flood.attempts.Load()), window)
		baseP50, floodP50 := baseLat.quantile(0.5), floodLat.quantile(0.5)
		r.check("legit replays all accepted during the flood", floodBad.Load() == 0 && baseBad.Load() == 0, "%d not ok alone, %d under flood, want 0", baseBad.Load(), floodBad.Load())
		r.check("legit median under forged flood stays near its quiet median (N1)", floodP50 <= 3*baseP50+50*time.Millisecond, "median %v quiet, %v under flood, want <= 3x quiet + 50ms", baseP50.Round(time.Millisecond), floodP50.Round(time.Millisecond))
	})

	registerTorture("webhooks-hang", "webhooks that never answer, sharing the delivery pool with healthy repos", func(r *tortureRun) {
		badRepos := r.load("bad_repos", 6)
		hooksPer := 4
		pushes := 40
		goodRepos := 3
		goodPushes := 30
		gap := 500 * time.Millisecond
		hang := r.duration("hang", 2*time.Minute)
		settle := r.duration("settle", 45*time.Second)

		w := newTortureWorld(r)
		r.gauge("webhook_rows", func() float64 { return w.rows("webhook_deliveries") })
		ks := w.addKnots(2)
		bad := w.addRepos(badRepos, func(int) (*tortureKnot, syntax.DID) { return ks[0], "did:plc:whbad" }, pushes, nil)
		good := w.addRepos(goodRepos, func(int) (*tortureKnot, syntax.DID) { return ks[1], "did:plc:whgood" }, goodPushes, nil)
		w.boot()
		w.seed(append(slices.Clone(bad), good...), 3*time.Minute)

		hung, fast := newAbuseHook(r, hang), newAbuseHook(r, 0)
		// webhook queues are per hostname, and localhost gets a different one from 127.0.0.1
		fastURL := strings.Replace(fast.srv.URL, "127.0.0.1", "localhost", 1)
		for _, repo := range bad {
			for i := range hooksPer {
				w.abuseAddHook(repo, fmt.Sprintf("%s/hook%d", hung.srv.URL, i))
			}
		}
		for _, repo := range good {
			w.abuseAddHook(repo, fastURL)
		}

		ctx := context.Background()
		var bst, gst deliveryStats
		var wg sync.WaitGroup
		wg.Go(func() { w.knotOutbox(ctx, abuseQueues(bad, 0, pushes)[ks[0]], &bst) })
		sent := map[string]time.Time{}
		wg.Go(func() {
			client := &http.Client{Timeout: time.Minute}
			for i := range goodPushes {
				for _, repo := range good {
					d := pushRange(repo, i, i+1)[0]
					sent[abuseHookKey(d)] = time.Now()
					reply := w.notifyWith(ctx, client, d)
					gst.saw(reply.status)
					gst.first.add(reply.elapsed)
				}
				sleepCtx(ctx, gap)
			}
		})
		wg.Wait()
		want := len(sent)
		r.waitFor(settle, func() bool { return int(fast.got.Load()) >= want })

		var hookLat latencies
		missing := 0
		for key, at := range sent {
			seen, ok := fast.seenAt(key)
			if !ok {
				missing++
				continue
			}
			hookLat.add(seen.Sub(at))
		}
		// otherwise a dropped delivery never shows up in the p99
		for range missing {
			hookLat.add(settle)
		}
		dids := make([]string, len(good))
		for i, repo := range good {
			dids[i] = "'" + repo.did.String() + "'"
		}
		saturated := w.countRows("select count(*) from webhook_deliveries where response_body = 'delivery queue saturated'")
		goodSaturated := w.countRows("select count(*) from webhook_deliveries d join webhooks h on h.id = d.webhook_id where d.response_body = 'delivery queue saturated' and h.repo_did in (" + strings.Join(dids, ",") + ")")
		hung.release()

		bst.report(r, "bad_notify_")
		gst.report(r, "good_notify_")
		r.metric("good_hooks_missing", float64(missing))
		r.metric("saturated_rows", saturated)
		r.metric("good_saturated_rows", goodSaturated)
		r.note("saturated rows are pruned to 50 per webhook, so the count is a floor")
		hookP99 := hookLat.quantile(0.99)
		pushP99 := gst.first.quantile(0.99)

		r.check("healthy webhooks all delivered while others hang (N7)", missing == 0, "%d of %d healthy deliveries missing %v after the last push, want 0", missing, want, settle)
		r.check("healthy webhooks delivered within 10s (N7)", hookP99 <= 10*time.Second, "p99 %v, want <= 10s", hookP99.Round(time.Millisecond))
		r.check("no healthy delivery dropped as queue saturated (N7)", goodSaturated == 0, "%v healthy deliveries recorded as saturated, want 0", goodSaturated)
		r.check("a push isn't slowed by the delivery backlog (N7)", pushP99 <= 3*time.Second, "healthy notify p99 %v, want <= 3s", pushP99.Round(time.Millisecond))
	})

	registerTorture("notify-hung-fetch", "one repo whose git fetch hangs while other repos get pushes", func(r *tortureRun) {
		victims := 3
		hold := r.duration("git_hold", 10*time.Second)
		budget := r.duration("victim_budget", 5*time.Second)

		w := newTortureWorld(r)
		r.gauge("notify_inflight", w.abuseNotifyInflight)
		ks := w.addKnots(2)
		hungKnot, okKnot := ks[0], ks[1]
		stuck := w.addRepo(hungKnot, "did:plc:hungowner", 2, nil)
		var others []*tortureRepo
		for range victims {
			others = append(others, w.addRepo(okKnot, "did:plc:hungok", 2, nil))
		}
		late := w.addRepo(okKnot, "did:plc:hungok", 0, nil)
		w.boot()
		w.seed(append([]*tortureRepo{stuck}, others...), 3*time.Minute)

		hungKnot.gitDelay.Store(int64(hold))
		ctx := context.Background()
		client := &http.Client{Timeout: 5 * time.Minute}
		defer client.CloseIdleConnections()
		var wg sync.WaitGroup
		var stuckReply notifyReply
		wg.Go(func() { stuckReply = w.notifyWith(ctx, client, pushRange(stuck, 0, 1)[0]) })
		if !r.waitFor(10*time.Second, func() bool { return hungKnot.fetches.Load() > 0 }) {
			r.t.Fatal("the stuck repo never reached git")
		}

		var otherLat latencies
		var notOK atomic.Int64
		for _, repo := range others {
			wg.Go(func() {
				reply := w.notifyWith(ctx, client, pushRange(repo, 0, 1)[0])
				otherLat.add(reply.elapsed)
				if reply.status != http.StatusOK {
					notOK.Add(1)
				}
			})
		}

		var refresh time.Duration
		var refreshErr error
		wg.Go(func() {
			start := time.Now()
			refreshErr = w.s.RefreshRepo(ctx, stuck.did)
			refresh = time.Since(start)
		})
		var lateServed time.Duration
		var lateOK bool
		wg.Go(func() {
			start := time.Now()
			w.emitDeclaration(late)
			w.emitManifest(late)
			lateOK = r.waitFor(3*time.Minute, func() bool { return w.servedCount([]*tortureRepo{late}) == 1 })
			lateServed = time.Since(start)
		})
		wg.Wait()

		r.latency("other_repo_notify", &otherLat)
		r.metric("stuck_notify_s", stuckReply.elapsed.Seconds())
		r.metric("stuck_repo_refresh_s", refresh.Seconds())
		r.metric("late_repo_ingest_s", lateServed.Seconds())
		r.note("git_hold %v is per git request, stuck_notify_s is how long the push really hung, production allows up to the 5 minute work budget", hold)

		r.check("the stuck repo's own push finishes", stuckReply.status == http.StatusOK, "answered %d: %s", stuckReply.status, stuckReply.body)
		r.check("all other repos' notifies answered ok", notOK.Load() == 0, "%d notifies not answered ok, want 0", notOK.Load())
		r.check("other repos aren't held up by a hung fetch (N4)", otherLat.quantile(1) <= budget, "slowest took %v, want <= %v", otherLat.quantile(1).Round(time.Millisecond), budget)
		r.check("a refresh of the repo isn't held up by its own hung fetch (N4)", refreshErr == nil && refresh <= budget, "refresh took %v with err %v, want <= %v", refresh.Round(time.Millisecond), refreshErr, budget)
		r.check("an unrelated repo is ingested while a fetch hangs (N4)", lateOK && lateServed <= 10*time.Second, "served=%v after %v, want served within 10s", lateOK, lateServed.Round(time.Millisecond))
	})
}