fix: harden record completion path

2026-06-10 18:24:30 +08:00
parent 4df04d1605
commit 13fd446f40
3 changed files with 212 additions and 5 deletions
@@ -0,0 +1,61 @@
+name: ci
+
+on:
+  push:
+    branches:
+      - main
+      - master
+      - codex/**
+    tags:
+      - v*
+  pull_request:
+  workflow_dispatch:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Go
+        uses: actions/setup-go@v5
+        with:
+          go-version-file: go.mod
+          cache: true
+
+      - name: Test
+        run: go test ./... -count=1
+
+      - name: Vet
+        run: go vet ./...
+
+  release-artifacts:
+    runs-on: ubuntu-latest
+    needs: test
+    if: startsWith(github.ref, 'refs/tags/') || github.event_name == 'workflow_dispatch'
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Go
+        uses: actions/setup-go@v5
+        with:
+          go-version-file: go.mod
+          cache: true
+
+      - name: Build release packages
+        run: |
+          VERSION="${GITHUB_REF_NAME:-dev}"
+          if [ "$VERSION" = "" ]; then
+            VERSION=dev
+          fi
+          make release VERSION="$VERSION"
+
+      - name: Upload release artifacts
+        uses: actions/upload-artifact@v4
+        with:
+          name: netstable-release
+          path: |
+            dist/*.tar.gz
+            dist/checksums.txt
@@ -186,6 +186,8 @@ type iperfSession struct {
 	BandwidthMbps float64
 }

+const maxCompleteBodyBytes = 64 * 1024 * 1024
+
 func New(options Options) *Server {
 	iperf3Path := strings.TrimSpace(options.Iperf3Path)
 	if iperf3Path == "" {
@@ -592,19 +594,18 @@ func (s *Server) handleCompleteIperf3Session(w http.ResponseWriter, r *http.Requ
 	}

 	var raw completeIperf3Request
-	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 2*1024*1024))
+	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxCompleteBodyBytes))
 	decoder.DisallowUnknownFields()
 	if err := decoder.Decode(&raw); err != nil {
 		writeJSON(w, http.StatusBadRequest, errorResponse{Error: "invalid JSON request"})
 		return
 	}

-	session, exists := s.takeIperf3Session(id)
+	session, exists := s.activeIperf3Session(id)
 	if !exists {
 		writeJSON(w, http.StatusNotFound, errorResponse{Error: "iperf3 session not found"})
 		return
 	}
-	stopIperf3Process(session)

 	summary := summarizeBrowserSamples(session.Request, completeTestRequest{
 		StartedAt:  raw.StartedAt,
@@ -623,6 +624,8 @@ func (s *Server) handleCompleteIperf3Session(w http.ResponseWriter, r *http.Requ
 			return
 		}
 	}
+	session, _ = s.takeIperf3Session(id)
+	stopIperf3Process(session)

 	writeJSON(w, http.StatusOK, completeTestResponse{Summary: summary})
 }
@@ -752,14 +755,14 @@ func (s *Server) handleCompleteTest(w http.ResponseWriter, r *http.Request) {
 	}

 	var raw completeTestRequest
-	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 2*1024*1024))
+	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxCompleteBodyBytes))
 	decoder.DisallowUnknownFields()
 	if err := decoder.Decode(&raw); err != nil {
 		writeJSON(w, http.StatusBadRequest, errorResponse{Error: "invalid JSON request"})
 		return
 	}

-	request, exists := s.takeSession(id)
+	request, exists := s.activeSession(id)
 	if !exists {
 		writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
 		return
@@ -778,6 +781,7 @@ func (s *Server) handleCompleteTest(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 	}
+	s.finishSession(id)

 	writeJSON(w, http.StatusOK, completeTestResponse{Summary: summary})
 }
@@ -965,6 +969,27 @@ func (s *Server) takeSession(id string) (TestRequest, bool) {
 	return request, exists
 }

+func (s *Server) activeSession(id string) (TestRequest, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.activeID != id {
+		return TestRequest{}, false
+	}
+	request, exists := s.sessions[id]
+	return request, exists
+}
+
+func (s *Server) finishSession(id string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.sessions, id)
+	s.removeQueuedLocked(id)
+	if s.activeID == id {
+		s.activeID = ""
+		s.promoteNextLocked(time.Now().UTC())
+	}
+}
+
 func (s *Server) takeIperf3Session(id string) (*iperfSession, bool) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -980,6 +1005,16 @@ func (s *Server) takeIperf3Session(id string) (*iperfSession, bool) {
 	return session, exists
 }

+func (s *Server) activeIperf3Session(id string) (*iperfSession, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.activeID != id {
+		return nil, false
+	}
+	session, exists := s.iperfSessions[id]
+	return session, exists
+}
+
 func (s *Server) expireIperf3Session(id string, ttl time.Duration) {
 	timer := time.NewTimer(ttl)
 	defer timer.Stop()
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"io/fs"
 	"net"
 	"net/http"
@@ -779,6 +780,69 @@ func TestCompleteEndpointPersistsBrowserSpeedRecordAndReleasesLock(t *testing.T)
 	}
 }

+func TestCompleteEndpointAcceptsLargeSamplePayload(t *testing.T) {
+	store := &memoryStore{}
+	server := New(Options{Store: store, Runner: noopRunner, ClientResolver: staticResolver})
+
+	createReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":30}`))
+	createReq.RemoteAddr = "203.0.113.20:54321"
+	createRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(createRes, createReq)
+	if createRes.Code != http.StatusAccepted {
+		t.Fatalf("create status = %d, want 202; body=%s", createRes.Code, createRes.Body.String())
+	}
+	var created createTestResponse
+	if err := json.Unmarshal(createRes.Body.Bytes(), &created); err != nil {
+		t.Fatalf("decode create response: %v", err)
+	}
+
+	completeReq := httptest.NewRequest(http.MethodPost, created.CompleteURL, largeCompleteBody(t, 30000))
+	completeRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(completeRes, completeReq)
+	if completeRes.Code != http.StatusOK {
+		t.Fatalf("complete status = %d, want 200 for large sample payload; body=%s", completeRes.Code, completeRes.Body.String())
+	}
+	records := store.snapshot()
+	if len(records) != 1 || len(records[0].Samples) != 30000 {
+		t.Fatalf("records = %d samples=%d, want one record with 30000 samples", len(records), len(records[0].Samples))
+	}
+}
+
+func TestCompleteEndpointCanRetryWhenPersistFails(t *testing.T) {
+	store := &flakyStore{fail: true}
+	server := New(Options{Store: store, Runner: noopRunner, ClientResolver: staticResolver})
+
+	createReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":30}`))
+	createReq.RemoteAddr = "203.0.113.20:54321"
+	createRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(createRes, createReq)
+	if createRes.Code != http.StatusAccepted {
+		t.Fatalf("create status = %d, want 202; body=%s", createRes.Code, createRes.Body.String())
+	}
+	var created createTestResponse
+	if err := json.Unmarshal(createRes.Body.Bytes(), &created); err != nil {
+		t.Fatalf("decode create response: %v", err)
+	}
+
+	firstComplete := httptest.NewRequest(http.MethodPost, created.CompleteURL, jsonBody(`{"samples":[{"kind":"download","success":true,"bytes":1024,"mbps":10}]}`))
+	firstRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(firstRes, firstComplete)
+	if firstRes.Code != http.StatusInternalServerError {
+		t.Fatalf("first complete status = %d, want 500; body=%s", firstRes.Code, firstRes.Body.String())
+	}
+
+	store.fail = false
+	secondComplete := httptest.NewRequest(http.MethodPost, created.CompleteURL, jsonBody(`{"samples":[{"kind":"download","success":true,"bytes":1024,"mbps":10}]}`))
+	secondRes := httptest.NewRecorder()
+	server.Handler().ServeHTTP(secondRes, secondComplete)
+	if secondRes.Code != http.StatusOK {
+		t.Fatalf("retry complete status = %d, want 200 after store recovers; body=%s", secondRes.Code, secondRes.Body.String())
+	}
+	if records := store.snapshot(); len(records) != 1 {
+		t.Fatalf("records = %d, want one record after retry", len(records))
+	}
+}
+
 func TestBrowserSummaryUsesSustainedTransferRatesForMax(t *testing.T) {
 	start := time.Date(2026, 6, 10, 8, 0, 0, 0, time.UTC)
 	request := TestRequest{
@@ -976,6 +1040,37 @@ func jsonBody(body string) *bytes.Buffer {
 	return bytes.NewBufferString(body)
 }

+func largeCompleteBody(t *testing.T, sampleCount int) *bytes.Buffer {
+	t.Helper()
+	var buffer bytes.Buffer
+	buffer.WriteString(`{"startedAt":"2026-06-05T08:00:00Z","finishedAt":"2026-06-05T08:00:30Z","samples":[`)
+	for i := 0; i < sampleCount; i++ {
+		if i > 0 {
+			buffer.WriteByte(',')
+		}
+		kind := "download"
+		if i%2 == 1 {
+			kind = "upload"
+		}
+		if err := json.NewEncoder(&buffer).Encode(probe.Sample{
+			At:        time.Date(2026, 6, 5, 8, 0, 0, 0, time.UTC).Add(time.Duration(i) * time.Millisecond),
+			Kind:      kind,
+			Success:   true,
+			Bytes:     4 * 1024 * 1024,
+			Mbps:      42.5,
+			LatencyMS: 150,
+		}); err != nil {
+			t.Fatalf("encode sample: %v", err)
+		}
+		data := buffer.Bytes()
+		if len(data) > 0 && data[len(data)-1] == '\n' {
+			buffer.Truncate(len(data) - 1)
+		}
+	}
+	buffer.WriteString(`]}`)
+	return &buffer
+}
+
 type memoryStore struct {
 	mu      sync.Mutex
 	records []recordstore.Record
@@ -1008,6 +1103,22 @@ func (s *memoryStore) snapshot() []recordstore.Record {
 	return out
 }

+type flakyStore struct {
+	memoryStore
+	mu   sync.Mutex
+	fail bool
+}
+
+func (s *flakyStore) Append(ctx context.Context, record recordstore.Record) error {
+	s.mu.Lock()
+	fail := s.fail
+	s.mu.Unlock()
+	if fail {
+		return errors.New("store unavailable")
+	}
+	return s.memoryStore.Append(ctx, record)
+}
+
 func noopRunner(ctx context.Context, request TestRequest, emit func(Event) error) (probe.Summary, error) {
 	return probe.Summary{Target: request.Target.Address, StartedAt: time.Now(), FinishedAt: time.Now()}, nil
 }