Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
124 changes: 52 additions & 72 deletions cmd/server/cdn_detection_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,6 @@ package main
// next.ServeHTTP.

import (
"bytes"
"log"
"net/http"
"net/http/httptest"
"strings"
Expand All @@ -40,22 +38,21 @@ func resetCDNDetectionOnce() {
// the middleware did not silently drop the request.
func runWithCDNMiddleware(t *testing.T, req *http.Request) (string, bool) {
t.Helper()
var buf bytes.Buffer
prev := log.Writer()
log.SetOutput(&buf)
defer log.SetOutput(prev)

nextCalled := false
h := cdnDetectionMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
nextCalled = true
w.WriteHeader(http.StatusOK)
}))
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("middleware must not block request; got status %d", w.Code)
}
return buf.String(), nextCalled
// The capture is locked (#310): goroutines from earlier tests keep
// logging into whatever writer is installed.
out := captureLog(func() {
h := cdnDetectionMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
nextCalled = true
w.WriteHeader(http.StatusOK)
}))
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("middleware must not block request; got status %d", w.Code)
}
})
return out, nextCalled
}

func TestCDNDetection_LogsOnCFRayHeader(t *testing.T) {
Expand Down Expand Up @@ -145,30 +142,29 @@ func TestCDNDetection_LogsWhenCDNHeaderAccompaniesProxyHeaders(t *testing.T) {
func TestCDNDetection_LogsOnlyOnce(t *testing.T) {
resetCDNDetectionOnce()

var buf bytes.Buffer
prev := log.Writer()
log.SetOutput(&buf)
defer log.SetOutput(prev)

nextCalled := 0
h := cdnDetectionMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
nextCalled++
w.WriteHeader(http.StatusOK)
}))

for i := 0; i < 3; i++ {
req := httptest.NewRequest("GET", "/api/observers", nil)
req.Header.Set("CF-Ray", "abc123")
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
}
// The capture is locked (#310): goroutines from earlier tests keep
// logging into whatever writer is installed.
out := captureLog(func() {
h := cdnDetectionMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
nextCalled++
w.WriteHeader(http.StatusOK)
}))

for i := 0; i < 3; i++ {
req := httptest.NewRequest("GET", "/api/observers", nil)
req.Header.Set("CF-Ray", "abc123")
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
}
})

if nextCalled != 3 {
t.Fatalf("middleware must call next on every request; got %d calls, want 3", nextCalled)
}
got := strings.Count(buf.String(), "detected request via CDN")
got := strings.Count(out, "detected request via CDN")
if got != 1 {
t.Errorf("expected CDN warning exactly once across multiple requests; got %d in output: %q", got, buf.String())
t.Errorf("expected CDN warning exactly once across multiple requests; got %d in output: %q", got, out)
}
}

Expand Down Expand Up @@ -206,56 +202,40 @@ func TestCDNDetection_RecognizesAllCommonCDNHeaders(t *testing.T) {
func TestCDNDetectionMiddlewareConcurrentFirstRequestLogsOnce(t *testing.T) {
resetCDNDetectionOnce()

var buf bytes.Buffer
var bufMu sync.Mutex
prev := log.Writer()
// log.Printf can be called concurrently; serialize writes to buf
// so we never race the test's own assertion read.
log.SetOutput(writerFunc(func(p []byte) (int, error) {
bufMu.Lock()
defer bufMu.Unlock()
return buf.Write(p)
}))
defer log.SetOutput(prev)

var nextCalls int64
h := cdnDetectionMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt64(&nextCalls, 1)
w.WriteHeader(http.StatusOK)
}))

const n = 50
var wg sync.WaitGroup
wg.Add(n)
for i := 0; i < n; i++ {
go func() {
defer wg.Done()
req := httptest.NewRequest("GET", "/api/observers", nil)
req.Header.Set("CF-Ray", "abc123-LAX")
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
}()
}
wg.Wait()
// log.Printf is called concurrently here; the shared capture is locked
// (#310), so this no longer needs a buffer and mutex of its own.
out := captureLog(func() {
h := cdnDetectionMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt64(&nextCalls, 1)
w.WriteHeader(http.StatusOK)
}))

var wg sync.WaitGroup
wg.Add(n)
for i := 0; i < n; i++ {
go func() {
defer wg.Done()
req := httptest.NewRequest("GET", "/api/observers", nil)
req.Header.Set("CF-Ray", "abc123-LAX")
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
}()
}
wg.Wait()
})

if got := atomic.LoadInt64(&nextCalls); got != n {
t.Fatalf("middleware must call next on every concurrent request; got %d, want %d", got, n)
}

bufMu.Lock()
out := buf.String()
bufMu.Unlock()
got := strings.Count(out, "detected request via CDN")
if got != 1 {
t.Errorf("expected sync.Once to admit exactly ONE warning under %d concurrent first-requests; got %d. Output:\n%s", n, got, out)
}
}

// writerFunc adapts a function to io.Writer.
type writerFunc func(p []byte) (int, error)

func (f writerFunc) Write(p []byte) (int, error) { return f(p) }

// Round-2 MAJOR finding: sync.Once only short-circuits the log.Printf,
// not the per-request header scan. firstCDNHeader still iterates 4
// http.Header.Get lookups on every /api request after warning fires.
Expand Down
68 changes: 58 additions & 10 deletions cmd/server/hash_migrate_log_race_301_test.go
Original file line number Diff line number Diff line change
@@ -1,20 +1,55 @@
package main

import (
"runtime"
"strings"
"sync"
"testing"
"time"
)

// hm301InstallSubpathGate holds the subpath index build until the returned
// release runs. release is idempotent, so the caller can defer it right after
// the gate goes in and still release it on purpose later: without that
// deferred net a t.Fatal between the two (store.Load returning an error, say)
// would leave the build goroutine blocked on the channel for the rest of the
// run (#310, nit 3 of the #307 review).
func hm301InstallSubpathGate(store *PacketStore) (release func()) {
gate := make(chan struct{})
store.subpathBuildGate = func() { <-gate }
return sync.OnceFunc(func() { close(gate) })
}

// The deferred net and the deliberate release are the same func, so releasing
// twice must be a no-op rather than a "close of closed channel" panic.
func TestHashMigrateSubpathGateReleaseIsIdempotent_310(t *testing.T) {
store := &PacketStore{}
release := hm301InstallSubpathGate(store)

blocked := make(chan struct{})
go func() {
defer close(blocked)
store.subpathBuildGate()
}()

release()
release()

select {
case <-blocked:
case <-time.After(30 * time.Second):
t.Fatal("the gated build goroutine was never released")
}
}

// Issue #301: Load() starts the two background index builds, and each builder
// logs one more line ("index build complete") AFTER it has flipped its ready
// flag. WaitIndexesReady returning therefore does not mean the builders are
// done logging, and a builder of an earlier store can log at any later time.
// The log capture of the hash-migrate harness must take those writes without a
// data race: here the subpath builder is released inside the capture, so its
// last line lands in the captured output while nothing orders it before the
// read at the end of the capture. Red under -race with an unsynchronized
// buffer.
// harness's reads. Red under -race with an unsynchronized buffer.
func TestHashMigrate_LogCaptureToleratesIndexBuilderLogs_301(t *testing.T) {
d := hm215Create(t)
d.ballast(t)
Expand All @@ -24,23 +59,36 @@ func TestHashMigrate_LogCaptureToleratesIndexBuilderLogs_301(t *testing.T) {
}
t.Cleanup(func() { db.conn.Close() })
store := NewPacketStore(db, &PacketStoreConfig{})
release := make(chan struct{})
store.subpathBuildGate = func() { <-release }
release := hm301InstallSubpathGate(store)
defer release() // a t.Fatal below must not strand the build goroutine
if err := store.Load(); err != nil {
t.Fatal(err)
}

out := hm215CaptureLog(func() {
close(release)
const want = "index build complete: subpath"
out := captureLogRead(func(read func() string) {
release()
if !store.WaitIndexesReady(30 * time.Second) {
t.Error("background index builds did not finish")
return
}
// The builder logs its completion line after the ready flag that
// woke us; give it time to land in the capture. Sleeping orders
// nothing, so the race detector still sees the two accesses.
time.Sleep(200 * time.Millisecond)
// woke us. Poll the capture for it rather than sleeping a fixed
// 200 ms (#310, nit 2 of the #307 review): the poll reads the
// buffer while the builder can still be writing to it and shares
// no lock with the builder's own progress, so an unsynchronized
// capture is still a race the detector reports — proven by the
// mutant in the PR report.
deadline := time.Now().Add(30 * time.Second)
for !strings.Contains(read(), want) {
if time.Now().After(deadline) {
t.Error("the subpath builder's last line did not arrive")
return
}
runtime.Gosched()
}
})
if !strings.Contains(out, "index build complete: subpath") {
if !strings.Contains(out, want) {
t.Fatalf("the subpath builder's last line was not captured:\n%s", out)
}
}
13 changes: 3 additions & 10 deletions cmd/server/hash_migrate_readonly_215_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ import (
"database/sql"
"database/sql/driver"
"fmt"
"log"
"os"
"path/filepath"
"reflect"
Expand Down Expand Up @@ -239,15 +238,9 @@ func hm215Migrate(s *PacketStore, batch int) string {
// hm215CaptureLog returns what the standard logger printed while fn ran. The
// buffer is locked: the index builders that Load() starts log their last line
// after the ready flag WaitIndexesReady waits for, so one can still be writing
// when the capture is read (#301).
func hm215CaptureLog(fn func()) string {
buf := newSyncBuffer()
prev := log.Writer()
log.SetOutput(buf)
defer log.SetOutput(prev)
fn()
return buf.String()
}
// when the capture is read (#301). Since #310 every cmd/server capture shares
// that one helper.
func hm215CaptureLog(fn func()) string { return captureLog(fn) }

// stmtLog records every statement a connection prepares or every transaction it
// begins. database/sql falls back to Prepare/Begin for a driver connection that
Expand Down
51 changes: 23 additions & 28 deletions cmd/server/issue199_node_not_found_test.go
Original file line number Diff line number Diff line change
@@ -1,10 +1,8 @@
package main

import (
"bytes"
"context"
"encoding/json"
"log"
"net/http/httptest"
"strings"
"testing"
Expand Down Expand Up @@ -167,18 +165,16 @@ func TestNodeDetail404LookupErrorIsLoggedOnce(t *testing.T) {
if _, err := srv.db.conn.Exec(`CREATE TABLE inactive_nodes (public_key TEXT PRIMARY KEY, name TEXT)`); err != nil {
t.Fatal(err)
}
var buf bytes.Buffer
prev := log.Writer()
log.SetOutput(&buf)
defer log.SetOutput(prev)

for i := 0; i < 3; i++ {
code, body := issue199Get(t, router, issue199Unknown)
if code != 404 || len(body) != 1 || body["error"] == nil {
t.Fatalf("request %d: status=%d body=%v, want the bare 404", i, code, body)
// The capture is locked (#310): the server's background goroutines keep
// logging into whatever writer is installed.
out := captureLog(func() {
for i := 0; i < 3; i++ {
code, body := issue199Get(t, router, issue199Unknown)
if code != 404 || len(body) != 1 || body["error"] == nil {
t.Fatalf("request %d: status=%d body=%v, want the bare 404", i, code, body)
}
}
}
out := buf.String()
})
if n := strings.Count(out, "missing-node lookup failed"); n != 1 {
t.Fatalf("logged %d lookup failures for 3 requests, want 1; log:\n%s", n, out)
}
Expand All @@ -193,21 +189,20 @@ func TestNodeDetail404LookupErrorIsLoggedOnce(t *testing.T) {
// A request cancelled by its client is not a broken lookup: nothing logged.
func TestNodeDetail404CancelledLookupIsNotLogged(t *testing.T) {
srv, _ := setupTestServer(t)
var buf bytes.Buffer
prev := log.Writer()
log.SetOutput(&buf)
defer log.SetOutput(prev)

ctx, cancel := context.WithCancel(context.Background())
cancel()
req := httptest.NewRequest("GET", "/api/nodes/"+issue199Unknown, nil).WithContext(ctx)
w := httptest.NewRecorder()
srv.writeNodeNotFound(w, req, issue199Unknown)
if w.Code != 404 {
t.Fatalf("status=%d, want 404", w.Code)
}
if strings.Contains(buf.String(), "missing-node lookup failed") {
t.Errorf("cancelled request logged as a lookup failure: %s", buf.String())
// The capture is locked (#310): the server's background goroutines keep
// logging into whatever writer is installed.
out := captureLog(func() {
ctx, cancel := context.WithCancel(context.Background())
cancel()
req := httptest.NewRequest("GET", "/api/nodes/"+issue199Unknown, nil).WithContext(ctx)
w := httptest.NewRecorder()
srv.writeNodeNotFound(w, req, issue199Unknown)
if w.Code != 404 {
t.Fatalf("status=%d, want 404", w.Code)
}
})
if strings.Contains(out, "missing-node lookup failed") {
t.Errorf("cancelled request logged as a lookup failure: %s", out)
}
}

Expand Down
Loading
Loading