EvoBGP/internal/httpapi/node_bundle_cache_test.go

package httpapi

import (
	"archive/tar"
	"bytes"
	"compress/gzip"
	"context"
	"crypto/ed25519"
	"encoding/base64"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"evobgp/internal/bundle"
	"evobgp/internal/signing"
	"evobgp/internal/store"
)

func TestNodeBundleCacheKeySeparatesTenantsRevisionsSpeakers(t *testing.T) {
	a := nodeBundleCacheKey("tenant-a", "rev", "spk", "overlay")
	if a == nodeBundleCacheKey("tenant-b", "rev", "spk", "overlay") {
		t.Fatal("tenants collided")
	}
	if a == nodeBundleCacheKey("tenant-a", "rev-2", "spk", "overlay") {
		t.Fatal("revisions collided")
	}
	if a == nodeBundleCacheKey("tenant-a", "rev", "spk-2", "overlay") {
		t.Fatal("speakers collided")
	}
	if a == nodeBundleCacheKey("tenant-a", "rev", "spk", "overlay-2") {
		t.Fatal("overlay inputs collided")
	}
}

func TestNodeBundleCacheHitSingleflightAndErrors(t *testing.T) {
	c := newNodeBundleCache(8, 1<<20)
	var builds atomic.Int32
	_, err := c.do(context.Background(), "k", func(context.Context) ([]byte, bool, error) {
		builds.Add(1)
		return []byte("signed-body"), true, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	got, err := c.do(context.Background(), "k", func(context.Context) ([]byte, bool, error) {
		builds.Add(1)
		return []byte("other"), true, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if builds.Load() != 1 {
		t.Fatalf("second pull rebuilt: builds=%d", builds.Load())
	}
	if !bytes.Equal(got, []byte("signed-body")) {
		t.Fatalf("hit body = %q", got)
	}

	t.Run("singleflight", func(t *testing.T) {
		c := newNodeBundleCache(8, 1<<20)
		var builds atomic.Int32
		release := make(chan struct{})
		const n = 8
		var wg sync.WaitGroup
		bodies := make([][]byte, n)
		errs := make([]error, n)
		wg.Add(n)
		for i := 0; i < n; i++ {
			go func(i int) {
				defer wg.Done()
				bodies[i], errs[i] = c.do(context.Background(), "same", func(context.Context) ([]byte, bool, error) {
					builds.Add(1)
					<-release
					return []byte("signed-bundle"), true, nil
				})
			}(i)
		}
		deadline := time.Now().Add(2 * time.Second)
		for {
			c.mu.Lock()
			joined := 0
			if f := c.flights["same"]; f != nil {
				joined = f.joined
			}
			c.mu.Unlock()
			if builds.Load() > 1 {
				close(release)
				wg.Wait()
				t.Fatalf("builds = %d before followers joined", builds.Load())
			}
			if builds.Load() == 1 && joined == n-1 {
				break
			}
			if time.Now().After(deadline) {
				close(release)
				wg.Wait()
				t.Fatalf("builds=%d joined=%d", builds.Load(), joined)
			}
			time.Sleep(time.Millisecond)
		}
		close(release)
		wg.Wait()
		if builds.Load() != 1 {
			t.Fatalf("builds = %d, want 1", builds.Load())
		}
		for i := 0; i < n; i++ {
			if errs[i] != nil {
				t.Fatalf("waiter %d: %v", i, errs[i])
			}
			if !bytes.Equal(bodies[i], []byte("signed-bundle")) {
				t.Fatalf("waiter %d body = %q", i, bodies[i])
			}
		}
	})

	t.Run("canceled waiter does not drop the build", func(t *testing.T) {
		c := newNodeBundleCache(4, 1024)
		release := make(chan struct{})
		var builds atomic.Int32
		leaderErr := make(chan error, 1)
		go func() {
			_, err := c.do(context.Background(), "k", func(context.Context) ([]byte, bool, error) {
				builds.Add(1)
				<-release
				return []byte("body"), true, nil
			})
			leaderErr <- err
		}()
		deadline := time.Now().Add(2 * time.Second)
		for {
			c.mu.Lock()
			_, ok := c.flights["k"]
			c.mu.Unlock()
			if ok || time.Now().After(deadline) {
				break
			}
			time.Sleep(time.Millisecond)
		}
		ctx, cancel := context.WithCancel(context.Background())
		cancel()
		_, err := c.do(ctx, "k", func(context.Context) ([]byte, bool, error) {
			builds.Add(1)
			return []byte("nope"), true, nil
		})
		if !errors.Is(err, context.Canceled) {
			t.Fatalf("follower err = %v", err)
		}
		close(release)
		if err := <-leaderErr; err != nil {
			t.Fatal(err)
		}
		if builds.Load() != 1 {
			t.Fatalf("builds = %d", builds.Load())
		}
		_, err = c.do(context.Background(), "k", func(context.Context) ([]byte, bool, error) {
			builds.Add(1)
			return []byte("other"), true, nil
		})
		if err != nil {
			t.Fatal(err)
		}
		if builds.Load() != 1 {
			t.Fatal("successful build was not cached after a canceled waiter")
		}
	})

	t.Run("errors are not cached", func(t *testing.T) {
		c := newNodeBundleCache(4, 1024)
		calls := 0
		boom := errors.New("httpapi: bundle build failed")
		_, err := c.do(context.Background(), "k", func(context.Context) ([]byte, bool, error) {
			calls++
			return nil, true, boom
		})
		if !errors.Is(err, boom) {
			t.Fatalf("err = %v", err)
		}
		_, err = c.do(context.Background(), "miss", func(context.Context) ([]byte, bool, error) {
			calls++
			return nil, false, errNodeBundleRevisionMissing
		})
		if !errors.Is(err, errNodeBundleRevisionMissing) {
			t.Fatalf("err = %v", err)
		}
		got, err := c.do(context.Background(), "k", func(context.Context) ([]byte, bool, error) {
			calls++
			return []byte("ok"), true, nil
		})
		if err != nil {
			t.Fatal(err)
		}
		if string(got) != "ok" || calls != 3 {
			t.Fatalf("calls=%d body=%q", calls, got)
		}
		_, err = c.do(context.Background(), "k", func(context.Context) ([]byte, bool, error) {
			calls++
			return []byte("other"), true, nil
		})
		if err != nil || calls != 3 {
			t.Fatalf("cached error path rebuilt: calls=%d err=%v", calls, err)
		}
	})

	t.Run("unstorable success is not cached", func(t *testing.T) {
		c := newNodeBundleCache(4, 1024)
		calls := 0
		for i := 0; i < 2; i++ {
			got, err := c.do(context.Background(), "k", func(context.Context) ([]byte, bool, error) {
				calls++
				return []byte("fresh"), false, nil
			})
			if err != nil || string(got) != "fresh" {
				t.Fatalf("i=%d err=%v body=%q", i, err, got)
			}
		}
		if calls != 2 {
			t.Fatalf("calls = %d, want 2", calls)
		}
	})
}

func TestNodeBundleCacheEvictsIdleAndRespectsCaps(t *testing.T) {
	t.Run("entry cap evicts oldest idle", func(t *testing.T) {
		c := newNodeBundleCache(2, 100)
		put := func(key, body string) {
			t.Helper()
			got, err := c.do(context.Background(), key, func(context.Context) ([]byte, bool, error) {
				return []byte(body), true, nil
			})
			if err != nil || string(got) != body {
				t.Fatalf("put %s: %v %q", key, err, got)
			}
		}
		put("a", "A")
		put("b", "B")
		if _, err := c.do(context.Background(), "a", func(context.Context) ([]byte, bool, error) {
			t.Fatal("recent entry was rebuilt")
			return nil, false, nil
		}); err != nil {
			t.Fatal(err)
		}
		put("c", "C")
		assertBundleCacheAccounting(t, c)
		if _, ok := c.items["b"]; ok {
			t.Fatal("idle entry b was kept")
		}
		if _, ok := c.items["a"]; !ok || c.items["c"] == nil {
			t.Fatalf("kept %v", cacheKeys(c))
		}
		if len(c.items) > 2 {
			t.Fatalf("entry cap exceeded: %d", len(c.items))
		}
	})

	t.Run("byte cap", func(t *testing.T) {
		c := newNodeBundleCache(8, 10)
		put := func(key string, n int) {
			t.Helper()
			body := bytes.Repeat([]byte("x"), n)
			got, err := c.do(context.Background(), key, func(context.Context) ([]byte, bool, error) {
				return body, true, nil
			})
			if err != nil || !bytes.Equal(got, body) {
				t.Fatalf("put %s: %v len=%d", key, err, len(got))
			}
			assertBundleCacheAccounting(t, c)
		}
		put("a", 4)
		put("b", 4)
		if _, err := c.do(context.Background(), "a", func(context.Context) ([]byte, bool, error) {
			t.Fatal("rebuilt a")
			return nil, false, nil
		}); err != nil {
			t.Fatal(err)
		}
		put("c", 4)
		assertBundleCacheAccounting(t, c)
		if _, ok := c.items["b"]; ok {
			t.Fatal("byte cap kept idle entry b")
		}
		if len(c.items) != 2 || c.total != 8 {
			t.Fatalf("n=%d total=%d", len(c.items), c.total)
		}
		put("d", 6)
		put("e", 6)
		assertBundleCacheAccounting(t, c)
		if len(c.items) != 1 || c.total != 6 {
			t.Fatalf("after oversized pair n=%d total=%d keys=%v", len(c.items), c.total, cacheKeys(c))
		}
	})

	t.Run("entry larger than the byte cap is not stored", func(t *testing.T) {
		c := newNodeBundleCache(4, 8)
		calls := 0
		big := bytes.Repeat([]byte("y"), 9)
		for i := 0; i < 2; i++ {
			got, err := c.do(context.Background(), "big", func(context.Context) ([]byte, bool, error) {
				calls++
				return big, true, nil
			})
			if err != nil || !bytes.Equal(got, big) {
				t.Fatalf("i=%d err=%v", i, err)
			}
		}
		if calls != 2 {
			t.Fatalf("oversize body was cached, calls=%d", calls)
		}
		if len(c.items) != 0 || c.total != 0 {
			t.Fatalf("stored oversize n=%d total=%d", len(c.items), c.total)
		}
	})
}

func TestNodeBundleOverlayFingerprintInputs(t *testing.T) {
	m := store.NewMemory()
	m.SeedDemo()
	tenant, _, _, rev, speaker := m.DemoIDs()

	base, err := nodeBundleOverlayFingerprint(m, tenant, speaker, rev)
	if err != nil {
		t.Fatal(err)
	}
	again, err := nodeBundleOverlayFingerprint(m, tenant, speaker, rev)
	if err != nil {
		t.Fatal(err)
	}
	if base != again {
		t.Fatal("fingerprint changed without an overlay input change")
	}

	sp, err := m.GetSpeaker(tenant, speaker)
	if err != nil {
		t.Fatal(err)
	}
	dispatch := store.MergeSpeakerMetaJSON(sp.MetaJSON, store.SpeakerMeta{
		LastDispatchAt:     "2026-01-01T00:00:00Z",
		LastDispatchStatus: "ok",
		SyncStatus:         "synced",
	})
	if _, err := m.UpdateSpeaker(tenant, speaker, &store.SpeakerPatch{MetaJSON: &dispatch}); err != nil {
		t.Fatal(err)
	}
	if got, err := nodeBundleOverlayFingerprint(m, tenant, speaker, rev); err != nil || got != base {
		t.Fatalf("dispatch meta changed fingerprint: %v %s", err, got)
	}

	peers := m.ListPeers(tenant)
	if len(peers) == 0 {
		t.Fatal("expected demo peers")
	}
	state := "Active"
	if _, err := m.UpdatePeer(tenant, peers[0].ID, &store.PeerPatch{SessionState: &state}); err != nil {
		t.Fatal(err)
	}
	if got, err := nodeBundleOverlayFingerprint(m, tenant, speaker, rev); err != nil || got != base {
		t.Fatalf("session state changed fingerprint: %v %s", err, got)
	}

	if err := m.PatchGlobalSettings(tenant, map[string]any{"bird_router_id": "203.0.113.7"}); err != nil {
		t.Fatal(err)
	}
	routed, err := nodeBundleOverlayFingerprint(m, tenant, speaker, rev)
	if err != nil {
		t.Fatal(err)
	}
	if routed == base {
		t.Fatal("router id setting did not change the overlay fingerprint")
	}

	nodeMeta := store.SpeakerMetaJSON(store.SpeakerMeta{NodeIPv4: "198.51.100.9"})
	if _, err := m.UpdateSpeaker(tenant, speaker, &store.SpeakerPatch{MetaJSON: &nodeMeta}); err != nil {
		t.Fatal(err)
	}
	nodeFP, err := nodeBundleOverlayFingerprint(m, tenant, speaker, rev)
	if err != nil {
		t.Fatal(err)
	}
	if nodeFP == routed {
		t.Fatal("node ipv4 did not change the overlay fingerprint")
	}

	sourceMeta := store.SpeakerMetaJSON(store.SpeakerMeta{
		NodeIPv4:          "198.51.100.9",
		BirdBgpSourceIPv4: "203.0.113.40",
	})
	if _, err := m.UpdateSpeaker(tenant, speaker, &store.SpeakerPatch{MetaJSON: &sourceMeta}); err != nil {
		t.Fatal(err)
	}
	sourceFP, err := nodeBundleOverlayFingerprint(m, tenant, speaker, rev)
	if err != nil {
		t.Fatal(err)
	}
	if sourceFP == nodeFP {
		t.Fatal("source ipv4 did not change the overlay fingerprint")
	}

	nodeOnly := store.SpeakerMetaJSON(store.SpeakerMeta{
		NodeIPv4:          "198.51.100.8",
		BirdBgpSourceIPv4: "203.0.113.40",
	})
	if _, err := m.UpdateSpeaker(tenant, speaker, &store.SpeakerPatch{MetaJSON: &nodeOnly}); err != nil {
		t.Fatal(err)
	}
	if got, err := nodeBundleOverlayFingerprint(m, tenant, speaker, rev); err != nil || got != sourceFP {
		t.Fatalf("unused node ipv4 changed fingerprint: %v %s vs %s", err, got, sourceFP)
	}

	v6 := store.SpeakerMetaJSON(store.SpeakerMeta{
		NodeIPv4:          "198.51.100.8",
		BirdBgpSourceIPv4: "203.0.113.40",
		BirdBgpSourceIPv6: "2001:db8::40",
	})
	if _, err := m.UpdateSpeaker(tenant, speaker, &store.SpeakerPatch{MetaJSON: &v6}); err != nil {
		t.Fatal(err)
	}
	v6FP, err := nodeBundleOverlayFingerprint(m, tenant, speaker, rev)
	if err != nil {
		t.Fatal(err)
	}
	// Source v6 changes local v6. When no peer policy overrides it, the rendered
	// peers fragment can stay the same. A difference means the overlay emitted it;
	// equality means a rebuild would not change the bundle.
	_ = v6FP

	if _, err := m.CreatePeer(tenant, &store.BGPPeer{
		Name:      "cache-extra",
		Neighbor:  "203.0.113.50",
		RemoteASN: 64512,
		Enabled:   true,
	}); err != nil {
		t.Fatal(err)
	}
	withPeer, err := nodeBundleOverlayFingerprint(m, tenant, speaker, rev)
	if err != nil {
		t.Fatal(err)
	}
	if withPeer == v6FP {
		t.Fatal("added peer did not change the overlay fingerprint")
	}

	sp2, err := m.CreateSpeaker(tenant, &store.Speaker{
		Role:     "replica",
		Endpoint: "https://203.0.113.2",
		MetaJSON: v6,
	})
	if err != nil {
		t.Fatal(err)
	}
	other, err := nodeBundleOverlayFingerprint(m, tenant, sp2.ID, rev)
	if err != nil {
		t.Fatal(err)
	}
	if other == withPeer {
		t.Fatal("speakers with the same addresses shared an overlay fingerprint")
	}

	otherRev, err := m.CreateRollbackRevision(tenant, rev)
	if err != nil {
		t.Fatal(err)
	}
	revFP, err := nodeBundleOverlayFingerprint(m, tenant, speaker, otherRev)
	if err != nil {
		t.Fatal(err)
	}
	if revFP == withPeer {
		t.Fatal("revision id did not change the overlay fingerprint")
	}
}

func TestHandleNodeBundleCache(t *testing.T) {
	srv, ts, ids := startNodeBundleServer(t)
	if srv.bundleCache.maxN != nodeBundleCacheMaxEntries || srv.bundleCache.maxBytes != nodeBundleCacheMaxBytes {
		t.Fatalf("caps n=%d bytes=%d", srv.bundleCache.maxN, srv.bundleCache.maxBytes)
	}
	var builds atomic.Int32
	srv.nodeBundleBuild = func(revisionID, speakerID string, fragments map[string]string, priv ed25519.PrivateKey) ([]byte, error) {
		builds.Add(1)
		return bundle.BuildGzippedTar(revisionID, speakerID, fragments, priv)
	}
	client := ts.Client()

	status, raw := pullNodeBundle(t, client, ts.URL, ids.speaker, "00000000-0000-0000-0000-000000000099")
	if status != http.StatusNotFound {
		t.Fatalf("missing revision status %d: %s", status, raw)
	}
	if builds.Load() != 0 {
		t.Fatal("missing revision built a bundle")
	}

	body1 := mustPullNodeBundle(t, client, ts.URL, ids.speaker, ids.rev)
	body2 := mustPullNodeBundle(t, client, ts.URL, ids.speaker, ids.rev)
	if builds.Load() != 1 {
		t.Fatalf("second pull rebuilt: builds=%d", builds.Load())
	}
	if !bytes.Equal(body1, body2) {
		t.Fatal("cache hit was not byte-for-byte identical")
	}
	verifyNodeBundle(t, srv, body2)

	sp2, err := srv.Store().CreateSpeaker(ids.tenant, &store.Speaker{
		Role:     "replica",
		Endpoint: "https://203.0.113.2",
		MetaJSON: store.SpeakerMetaJSON(store.SpeakerMeta{NodeIPv4: "203.0.113.2"}),
	})
	if err != nil {
		t.Fatal(err)
	}
	bodyB := mustPullNodeBundle(t, client, ts.URL, sp2.ID, ids.rev)
	if builds.Load() != 2 {
		t.Fatalf("second speaker builds=%d", builds.Load())
	}
	if bytes.Equal(body1, bodyB) {
		t.Fatal("different speakers returned the same bundle")
	}
	if !strings.Contains(bundleFile(t, bodyB, "bird.conf"), "203.0.113.2") {
		t.Fatal("speaker B bundle missing its router id")
	}
	body3 := mustPullNodeBundle(t, client, ts.URL, ids.speaker, ids.rev)
	if builds.Load() != 2 || !bytes.Equal(body1, body3) {
		t.Fatalf("speaker A collided with B: builds=%d equal=%v", builds.Load(), bytes.Equal(body1, body3))
	}

	meta := store.SpeakerMetaJSON(store.SpeakerMeta{NodeIPv4: "198.51.100.9"})
	if _, err := srv.Store().UpdateSpeaker(ids.tenant, ids.speaker, &store.SpeakerPatch{MetaJSON: &meta}); err != nil {
		t.Fatal(err)
	}
	body4 := mustPullNodeBundle(t, client, ts.URL, ids.speaker, ids.rev)
	if builds.Load() != 3 {
		t.Fatalf("changed node ipv4 builds=%d, want 3", builds.Load())
	}
	if bytes.Equal(body1, body4) {
		t.Fatal("changed overlay input reused the previous bundle")
	}
	conf := bundleFile(t, body4, "bird.conf")
	if !strings.Contains(conf, "198.51.100.9") {
		t.Fatalf("rebuilt bird.conf missing node ipv4: %s", conf)
	}
	verifyNodeBundle(t, srv, body4)
	body5 := mustPullNodeBundle(t, client, ts.URL, ids.speaker, ids.rev)
	if builds.Load() != 3 || !bytes.Equal(body4, body5) {
		t.Fatal("rebuilt bundle was not cached")
	}

	source := store.SpeakerMetaJSON(store.SpeakerMeta{
		NodeIPv4:          "198.51.100.9",
		BirdBgpSourceIPv4: "203.0.113.40",
	})
	if _, err := srv.Store().UpdateSpeaker(ids.tenant, ids.speaker, &store.SpeakerPatch{MetaJSON: &source}); err != nil {
		t.Fatal(err)
	}
	body6 := mustPullNodeBundle(t, client, ts.URL, ids.speaker, ids.rev)
	if builds.Load() != 4 {
		t.Fatalf("changed source ipv4 builds=%d, want 4", builds.Load())
	}
	conf = bundleFile(t, body6, "bird.conf")
	if !strings.Contains(conf, "203.0.113.40") {
		t.Fatalf("source ipv4 not applied: %s", conf)
	}
	verifyNodeBundle(t, srv, body6)

	otherRev, err := srv.Store().CreateRollbackRevision(ids.tenant, ids.rev)
	if err != nil {
		t.Fatal(err)
	}
	bodyRev := mustPullNodeBundle(t, client, ts.URL, ids.speaker, otherRev)
	if builds.Load() != 5 {
		t.Fatalf("other revision builds=%d", builds.Load())
	}
	if bytes.Equal(body6, bodyRev) {
		t.Fatal("revisions shared a bundle body")
	}
	cached := mustPullNodeBundle(t, client, ts.URL, ids.speaker, ids.rev)
	if builds.Load() != 5 || !bytes.Equal(body6, cached) {
		t.Fatal("original revision was evicted or rebuilt")
	}
}

func TestHandleNodeBundleSingleflight(t *testing.T) {
	srv, ts, ids := startNodeBundleServer(t)
	var builds atomic.Int32
	release := make(chan struct{})
	srv.nodeBundleBuild = func(revisionID, speakerID string, fragments map[string]string, priv ed25519.PrivateKey) ([]byte, error) {
		builds.Add(1)
		<-release
		return bundle.BuildGzippedTar(revisionID, speakerID, fragments, priv)
	}
	client := ts.Client()
	const n = 2
	bodies := make([][]byte, n)
	errs := make([]error, n)
	var wg sync.WaitGroup
	wg.Add(n)
	for i := 0; i < n; i++ {
		go func(i int) {
			defer wg.Done()
			bodies[i], errs[i] = pullNodeBundleOK(client, ts.URL, ids.speaker, ids.rev)
		}(i)
	}
	deadline := time.Now().Add(3 * time.Second)
	joined := 0
	for {
		if builds.Load() > 1 {
			close(release)
			wg.Wait()
			t.Fatalf("concurrent pulls built %d times", builds.Load())
		}
		srv.bundleCache.mu.Lock()
		joined = 0
		for _, f := range srv.bundleCache.flights {
			joined += f.joined
		}
		srv.bundleCache.mu.Unlock()
		if builds.Load() == 1 && joined >= 1 {
			break
		}
		if time.Now().After(deadline) {
			close(release)
			wg.Wait()
			t.Fatalf("timed out builds=%d joined=%d", builds.Load(), joined)
		}
		time.Sleep(time.Millisecond)
	}
	close(release)
	wg.Wait()
	if builds.Load() != 1 {
		t.Fatalf("builds = %d, want 1", builds.Load())
	}
	for i := 0; i < n; i++ {
		if errs[i] != nil {
			t.Fatal(errs[i])
		}
	}
	if !bytes.Equal(bodies[0], bodies[1]) {
		t.Fatal("singleflight returned different bodies")
	}
	verifyNodeBundle(t, srv, bodies[0])
}

type nodeBundleIDs struct {
	tenant, rev, speaker string
}

func startNodeBundleServer(t *testing.T) (*Server, *httptest.Server, nodeBundleIDs) {
	t.Helper()
	srv, err := New(Options{
		InsecureDev:   true,
		SeedDemo:      true,
		BundleSeedHex: testBundleSeed,
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(srv.Close)
	tenant, _, _, rev, speaker := srv.Store().DemoIDs()
	mustSetTestAPIKeys(t, srv, "nodekey|"+tenant+"|node")
	ts := httptest.NewServer(srv.Handler())
	t.Cleanup(ts.Close)
	return srv, ts, nodeBundleIDs{tenant: tenant, rev: rev, speaker: speaker}
}

func mustPullNodeBundle(t *testing.T, client *http.Client, base, speaker, rev string) []byte {
	t.Helper()
	status, body := pullNodeBundle(t, client, base, speaker, rev)
	if status != http.StatusOK {
		t.Fatalf("status %d: %s", status, body)
	}
	return body
}

func pullNodeBundle(t *testing.T, client *http.Client, base, speaker, rev string) (int, []byte) {
	t.Helper()
	req, err := http.NewRequest(http.MethodGet, base+"/v1/speakers/"+speaker+"/bundle/"+rev, nil)
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("Authorization", "Bearer nodekey")
	resp, err := client.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		t.Fatal(err)
	}
	return resp.StatusCode, body
}

func pullNodeBundleOK(client *http.Client, base, speaker, rev string) ([]byte, error) {
	req, err := http.NewRequest(http.MethodGet, base+"/v1/speakers/"+speaker+"/bundle/"+rev, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("Authorization", "Bearer nodekey")
	resp, err := client.Do(req)
	if err != nil {
		return nil, err
	}
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	if resp.StatusCode != http.StatusOK {
		return nil, errors.New(resp.Status + ": " + string(body))
	}
	return body, nil
}

func verifyNodeBundle(t *testing.T, srv *Server, raw []byte) {
	t.Helper()
	pub, err := base64.StdEncoding.DecodeString(srv.BundlePublicKeyBase64())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := signing.VerifyGzippedTar(raw, ed25519.PublicKey(pub)); err != nil {
		t.Fatal(err)
	}
}

func bundleFile(t *testing.T, gz []byte, name string) string {
	t.Helper()
	gr, err := gzip.NewReader(bytes.NewReader(gz))
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = gr.Close() }()
	tr := tar.NewReader(gr)
	for {
		hdr, err := tr.Next()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			t.Fatal(err)
		}
		if hdr.Name != name {
			continue
		}
		var b strings.Builder
		if _, err := io.Copy(&b, tr); err != nil {
			t.Fatal(err)
		}
		return b.String()
	}
	t.Fatalf("archive missing %s", name)
	return ""
}

func assertBundleCacheAccounting(t *testing.T, c *nodeBundleCache) {
	t.Helper()
	if len(c.items) != len(c.lru) {
		t.Fatalf("items %d lru %d", len(c.items), len(c.lru))
	}
	sum := 0
	seen := make(map[string]bool, len(c.lru))
	for _, k := range c.lru {
		e, ok := c.items[k]
		if !ok {
			t.Fatalf("lru key %q missing from items", k)
		}
		if seen[k] {
			t.Fatalf("duplicate lru key %q", k)
		}
		seen[k] = true
		sum += len(e.body)
	}
	if sum != c.total {
		t.Fatalf("total=%d sum=%d", c.total, sum)
	}
	if c.maxN > 0 && len(c.items) > c.maxN {
		t.Fatalf("entries %d over cap %d", len(c.items), c.maxN)
	}
	if c.maxBytes > 0 && c.total > c.maxBytes {
		t.Fatalf("bytes %d over cap %d", c.total, c.maxBytes)
	}
}

func cacheKeys(c *nodeBundleCache) []string {
	keys := make([]string, 0, len(c.items))
	for k := range c.items {
		keys = append(keys, k)
	}
	return keys
}