Cyrene_v2/internal/services/gateway/observability_metrics_test.go

package gateway

import (
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"sync"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
)

// These tests pin the contract that broke in production (2026-10-03 21:00:28):
// MetricsCollector.Snapshot must hand back DETACHED counter maps, because the
// admin handler JSON-encodes the snapshot after the collector's lock is released.
// Returning the collector's own maps let a concurrent Record write a map while the
// encoder iterated it → unrecoverable
// "fatal error: concurrent map iteration and map write" → the whole backend died,
// and with it every admin page (the reported「全链路流程功能无效」).

// TestMetricsSnapshotDetachesMaps is the deterministic regression: a value read out
// of a snapshot must not change when the collector is counted afterwards, and
// writing into a snapshot must not reach the collector.
func TestMetricsSnapshotDetachesMaps(t *testing.T) {
	t.Parallel()
	m := newMetricsCollector()
	m.Record(http.MethodGet, "/api/v1/a", http.StatusOK, 1)

	snap := m.Snapshot(time.Now())
	if got := snap.PerPath["/api/v1/a"]; got != 1 {
		t.Fatalf("PerPath[/api/v1/a] = %d, want 1", got)
	}
	if got := snap.PerMethod[http.MethodGet]; got != 1 {
		t.Fatalf("PerMethod[GET] = %d, want 1", got)
	}

	// A count recorded after the snapshot must not appear in the older snapshot.
	m.Record(http.MethodPost, "/api/v1/b", http.StatusTeapot, 1)
	if _, ok := snap.PerPath["/api/v1/b"]; ok {
		t.Fatalf("snapshot shares the collector's perPath map: later path leaked into it (%v)", snap.PerPath)
	}
	if _, ok := snap.PerMethod[http.MethodPost]; ok {
		t.Fatalf("snapshot shares the collector's perMethod map: later method leaked into it (%v)", snap.PerMethod)
	}
	if _, ok := snap.StatusCounts["418"]; ok {
		t.Fatalf("snapshot shares the collector's status map: later status leaked into it (%v)", snap.StatusCounts)
	}

	assertSnapshotWriteIsolation(t, m, snap)
	assertCollectorNotMutatedBySnapshot(t, m)
}

// assertSnapshotWriteIsolation writes into the snapshot's maps and checks the
// collector is untouched afterwards.
func assertSnapshotWriteIsolation(t *testing.T, m *MetricsCollector, snap AdminMetrics) {
	t.Helper()
	snap.PerPath["/api/v1/evil"] = 999
	snap.PerMethod["DELETE"] = 999
	snap.StatusCounts["500"] = 999

	after := m.Snapshot(time.Now())
	if _, ok := after.PerPath["/api/v1/evil"]; ok {
		t.Fatalf("collector's perPath is the snapshot's map: %v", after.PerPath)
	}
	if _, ok := after.PerMethod["DELETE"]; ok {
		t.Fatalf("collector's perMethod is the snapshot's map: %v", after.PerMethod)
	}
	if _, ok := after.StatusCounts["500"]; ok {
		t.Fatalf("collector's status map is the snapshot's map: %v", after.StatusCounts)
	}
	if after.TotalRequests != 2 {
		t.Fatalf("TotalRequests = %d, want 2", after.TotalRequests)
	}
	if after.PerPath["/api/v1/b"] != 1 {
		t.Fatalf("PerPath[/api/v1/b] = %d, want 1 (real count lost)", after.PerPath["/api/v1/b"])
	}
}

// assertCollectorNotMutatedBySnapshot re-reads a counter the collector recorded
// before the isolation writes to prove no counter value was disturbed.
func assertCollectorNotMutatedBySnapshot(t *testing.T, m *MetricsCollector) {
	t.Helper()
	final := m.Snapshot(time.Now())
	if final.PerPath["/api/v1/a"] != 1 {
		t.Fatalf("PerPath[/api/v1/a] = %d, want 1", final.PerPath["/api/v1/a"])
	}
	if final.PerMethod[http.MethodGet] != 1 {
		t.Fatalf("PerMethod[GET] = %d, want 1", final.PerMethod[http.MethodGet])
	}
}

// TestMetricsSnapshotEncodedConcurrently runs the exact interleaving that crashed
// production: one goroutine counts requests (the logging middleware does this on
// every request) while another encodes snapshots (the admin handler does this
// outside the lock). Under -race a shared live map is reported as a data race; on
// an unpatched build it can also abort the process outright.
func TestMetricsSnapshotEncodedConcurrently(t *testing.T) {
	t.Parallel()
	m := newMetricsCollector()
	stop := make(chan struct{})
	var wg sync.WaitGroup

	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case <-stop:
				return
			default:
			}
			m.Record(http.MethodGet, "/api/v1/x", http.StatusOK, 1)
		}
	}()

	deadline := time.Now().Add(300 * time.Millisecond)
	for time.Now().Before(deadline) {
		if _, err := json.Marshal(m.Snapshot(time.Now())); err != nil {
			close(stop)
			wg.Wait()
			t.Fatalf("marshal snapshot: %v", err)
		}
	}
	close(stop)
	wg.Wait()
}

// TestAdminMetricsEndpointConcurrentPoll drives the real HTTP path — the admin
// panel polls every module in parallel, so one request can be encoding the metrics
// JSON while another request's logging middleware records into the collector. This
// is the production crash reproduced end-to-end through the middleware + handler.
func TestAdminMetricsEndpointConcurrentPoll(t *testing.T) {
	env := newEnv(t, nil, false)
	tok := env.token(t, auth.RoleOwner)
	metricsURL := env.ts.URL + "/api/v1/admin/observability/metrics"
	healthURL := env.ts.URL + "/api/v1/health"

	// Both jobs run concurrently: the panel's metrics reads race the load requests
	// that keep recording into the collector.
	for _, err := range runConcurrent(
		func() error { return driveRequests(healthURL, "", 4, 25) },
		func() error { return driveRequests(metricsURL, tok, 6, 25) },
	) {
		if err != nil {
			t.Fatalf("concurrent metrics poll: %v", err)
		}
	}

	resp, body := doReq(t, http.MethodGet, metricsURL, nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	var payload struct {
		Data struct {
			TotalRequests int64            `json:"totalRequests"`
			PerPath       map[string]int64 `json:"perPath"`
		} `json:"data"`
	}
	if err := json.Unmarshal(body, &payload); err != nil {
		t.Fatalf("decode metrics: %v (body=%s)", err, body)
	}
	if payload.Data.TotalRequests == 0 {
		t.Fatal("totalRequests = 0, want the load above to be counted")
	}
	if payload.Data.PerPath["/api/v1/health"] == 0 {
		t.Fatalf("perPath[/api/v1/health] = 0, want the health load counted (%v)", payload.Data.PerPath)
	}
}

// runConcurrent runs every job in its own goroutine and returns the non-nil errors.
func runConcurrent(jobs ...func() error) []error {
	var wg sync.WaitGroup
	errs := make([]error, len(jobs))
	for i, job := range jobs {
		wg.Add(1)
		go func(i int, job func() error) {
			defer wg.Done()
			errs[i] = job()
		}(i, job)
	}
	wg.Wait()
	return errs
}

// driveRequests issues n GETs of url, workers of them in parallel, using token as
// a bearer credential when non-empty. It returns the first failure (status or I/O).
func driveRequests(url, token string, workers, n int) error {
	var wg sync.WaitGroup
	fail := make(chan error, workers)

	for w := 0; w < workers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := 0; i < n; i++ {
				req, err := http.NewRequest(http.MethodGet, url, nil)
				if err != nil {
					select {
					case fail <- err:
					default:
					}
					return
				}
				if token != "" {
					req.Header.Set("Authorization", "Bearer "+token)
				}
				resp, err := http.DefaultClient.Do(req)
				if err != nil {
					select {
					case fail <- err:
					default:
					}
					return
				}
				body, err := io.ReadAll(resp.Body)
				_ = resp.Body.Close()
				if err != nil {
					select {
					case fail <- err:
					default:
					}
					return
				}
				if resp.StatusCode != http.StatusOK {
					select {
					case fail <- fmt.Errorf("GET %s status = %d (body=%s)", url, resp.StatusCode, body):
					default:
					}
					return
				}
			}
		}()
	}
	wg.Wait()
	close(fail)
	for err := range fail {
		return err
	}
	return nil
}