From 13fd446f408f572c75639bffe083b7c7b39c7eb6 Mon Sep 17 00:00:00 2001 From: Test User Date: Wed, 10 Jun 2026 18:24:30 +0800 Subject: [PATCH] fix: harden record completion path --- .gitea/workflows/ci.yml | 61 ++++++++++++++++++++ internal/web/server.go | 45 +++++++++++++-- internal/web/server_test.go | 111 ++++++++++++++++++++++++++++++++++++ 3 files changed, 212 insertions(+), 5 deletions(-) create mode 100644 .gitea/workflows/ci.yml diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..f47d883 --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -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 diff --git a/internal/web/server.go b/internal/web/server.go index 8d47bab..1fe9b9f 100644 --- a/internal/web/server.go +++ b/internal/web/server.go @@ -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() diff --git a/internal/web/server_test.go b/internal/web/server_test.go index b397bff..a4ab2ea 100644 --- a/internal/web/server_test.go +++ b/internal/web/server_test.go @@ -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 }