refactor: fethcer interface & download flow (#1229)

This commit is contained in:
Levi authored and GitHub committed 2026-01-08 21:50:47 +08:00
1 parent abd16d5326
commit a9eee26be7
30 files changed
+2961 -1631

No files matched your search

+210 -28
View File
@@ -70,6 +70,9 @@ func StartTestSlowFileServer(delay time.Duration) net.Listener {
func StartTestCustomServer() net.Listener {
return startTestServer(func(sl *shutdownListener) http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/", func(writer http.ResponseWriter, request *http.Request) {
writer.WriteHeader(200)
})
mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) {
file, err := os.Open(BuildFile)
if err != nil {
@@ -221,6 +224,10 @@ func StartTestCustomServer() net.Listener {
defer file.Close()
io.Copy(writer, file)
})
// Test 403 Forbidden endpoint
mux.HandleFunc("/forbidden", func(writer http.ResponseWriter, request *http.Request) {
writer.WriteHeader(403)
})
return mux
})
}
@@ -244,20 +251,6 @@ func StartTestHostHeaderServer() net.Listener {
})
}
// StartTestRootServer starts a simple server at the root path
// Used to test URL resolution when no filename is provided
func StartTestRootServer() net.Listener {
return startTestServer(func(sl *shutdownListener) http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/", func(writer http.ResponseWriter, request *http.Request) {
writer.Header().Set("Content-Type", "text/html")
writer.WriteHeader(200)
writer.Write([]byte("<html><body>Test Page</body></html>"))
})
return mux
})
}
func StartTestRetryServer() net.Listener {
counter := 0
return startTestServer(func(sl *shutdownListener) http.Handler {
@@ -303,23 +296,100 @@ func StartTestPostServer() net.Listener {
}
func StartTestErrorServer() net.Listener {
counter := 0
return startTestServer(func(sl *shutdownListener) http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) {
counter++
if counter != 1 {
writer.WriteHeader(500)
return
}
// Always return 404 error to test error handling
writer.WriteHeader(404)
return
})
return mux
})
}
// StartTestOneTimeServer creates a server where the URL can only be downloaded once
// The first non-probe request succeeds, all subsequent requests return 404
// This simulates one-time download URLs (e.g., signed URLs that expire after first use)
func StartTestOneTimeServer() net.Listener {
var accessed atomic.Bool
return startTestServer(func(sl *shutdownListener) http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) {
// First full download request succeeds, subsequent requests fail
if accessed.Swap(true) {
writer.WriteHeader(404)
return
}
// First request - return full file without Range support
writer.Header().Set("Content-Length", fmt.Sprintf("%d", BuildSize))
file, err := os.Open(BuildFile)
if err != nil {
return
}
defer file.Close()
io.Copy(writer, file)
})
return mux
})
}
// StartTestSlowStartServer creates a server with configurable delay per request
// This allows testing slow-start connection expansion to reach max connections
func StartTestSlowStartServer(delayPerByte time.Duration) net.Listener {
return startTestServer(func(sl *shutdownListener) http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) {
rangeFileHandle(
writer,
request,
nil,
func(file *os.File, n int64) {
slowCopyNWithDelay(sl, writer, file, n, delayPerByte)
},
)
})
return mux
})
}
// slowCopyNWithDelay copies n bytes from src to dst with a delay per byte
func slowCopyNWithDelay(sl *shutdownListener, dst io.Writer, src io.Reader, n int64, delayPerByte time.Duration) {
buf := make([]byte, 32*1024)
remaining := n
for remaining > 0 {
if sl.isShutdown {
return
}
toRead := int64(len(buf))
if toRead > remaining {
toRead = remaining
}
nr, er := src.Read(buf[:toRead])
if nr > 0 {
nw, ew := dst.Write(buf[0:nr])
if nw > 0 {
remaining -= int64(nw)
// Add delay based on bytes written
if delayPerByte > 0 {
time.Sleep(delayPerByte * time.Duration(nw))
}
}
if ew != nil {
break
}
}
if er != nil {
break
}
}
}
// StartTestLimitServer connections limit server
func StartTestLimitServer(maxConnections int32, delay int64) net.Listener {
var connections atomic.Int32
var slowOnce atomic.Bool
return startTestServer(func(sl *shutdownListener) http.Handler {
mux := http.NewServeMux()
@@ -332,12 +402,93 @@ func StartTestLimitServer(maxConnections int32, delay int64) net.Listener {
writer.WriteHeader(403)
return
}
// First request intentionally delays the first write to trigger a single read timeout,
// subsequent requests respond at normal speed.
useInitialDelay := delay > 0 && !slowOnce.Swap(true)
rangeFileHandle(
writer,
request,
nil,
func(file *os.File, n int64) {
slowCopyN(sl, writer, file, n, delay)
if useInitialDelay {
slowCopyAfterDelay(sl, writer, file, n, delay)
return
}
slowCopyN(sl, writer, file, n, 0)
},
)
})
return mux
})
}
// StartTestTimeoutOnceServer creates a server that times out on first request, then works normally
func StartTestTimeoutOnceServer(delay int64) net.Listener {
var timeoutOnce atomic.Bool
return startTestServer(func(sl *shutdownListener) http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) {
// First request delays to trigger timeout, subsequent requests respond normally
useInitialDelay := delay > 0 && !timeoutOnce.Swap(true)
rangeFileHandle(
writer,
request,
nil,
func(file *os.File, n int64) {
if useInitialDelay {
slowCopyAfterDelay(sl, writer, file, n, delay)
return
}
slowCopyN(sl, writer, file, n, 0)
},
)
})
return mux
})
}
// StartTestTemporary500Server creates a server that returns 500 for a duration, then recovers
// Uses slow transfer to ensure the file isn't fully downloaded during resolve phase
func StartTestTemporary500Server(errorDuration time.Duration) net.Listener {
startTime := time.Now()
var requestCount atomic.Int32
return startTestServer(func(sl *shutdownListener) http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) {
reqNum := requestCount.Add(1)
// First request (Resolve) succeeds with slow transfer to prevent full download
// Subsequent requests return 500 for the specified duration, then recover
if reqNum == 1 {
// Slow transfer for resolve: 50 microsecond per byte (~20MB/s)
// This ensures resolve doesn't complete the full 200MB file
rangeFileHandle(
writer,
request,
nil,
func(file *os.File, n int64) {
slowCopyNWithDelay(sl, writer, file, n, 50*time.Microsecond)
},
)
return
}
// Subsequent requests: return 500 for errorDuration, then normal
if time.Since(startTime) < errorDuration {
writer.WriteHeader(500)
writer.Write([]byte("Internal Server Error"))
return
}
rangeFileHandle(
writer,
request,
nil,
func(file *os.File, n int64) {
slowCopyN(sl, writer, file, n, 0)
},
)
})
@@ -375,6 +526,26 @@ func StartTestRangeBugServer() net.Listener {
func rangeFileHandle(writer http.ResponseWriter, request *http.Request, modifyEnd func(end int64) int64, iocpN func(file *os.File, n int64)) {
r := request.Header.Get("Range")
// If no Range header, return full file with Accept-Ranges header
if r == "" {
// Open file first to ensure it exists
file, err := os.Open(BuildFile)
if err != nil {
writer.WriteHeader(500)
return
}
defer file.Close()
writer.Header().Set("Content-Length", fmt.Sprintf("%d", BuildSize))
writer.Header().Set("Accept-Ranges", "bytes")
writer.WriteHeader(200)
(writer.(http.Flusher)).Flush()
iocpN(file, BuildSize)
return
}
// split range
s := strings.Split(r, "=")
if len(s) != 2 {
@@ -408,12 +579,7 @@ func rangeFileHandle(writer http.ResponseWriter, request *http.Request, modifyEn
end = modifyEnd(end)
}
writer.Header().Set("Content-Length", fmt.Sprintf("%d", end-start+1))
writer.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, BuildSize))
writer.Header().Set("Accept-Ranges", "bytes")
writer.WriteHeader(206)
(writer.(http.Flusher)).Flush()
// Open file before sending headers to ensure it exists
file, err := os.Open(BuildFile)
if err != nil {
writer.WriteHeader(500)
@@ -421,6 +587,13 @@ func rangeFileHandle(writer http.ResponseWriter, request *http.Request, modifyEn
}
defer file.Close()
file.Seek(start, 0)
writer.Header().Set("Content-Length", fmt.Sprintf("%d", end-start+1))
writer.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, BuildSize))
writer.Header().Set("Accept-Ranges", "bytes")
writer.WriteHeader(206)
(writer.(http.Flusher)).Flush()
iocpN(file, end-start+1)
}
@@ -471,6 +644,14 @@ func slowCopyN(sl *shutdownListener, dst io.Writer, src io.Reader, n int64, dela
return
}
// slowCopyAfterDelay sleeps once before performing a normal copy to simulate a single timeout.
func slowCopyAfterDelay(sl *shutdownListener, dst io.Writer, src io.Reader, n int64, delay int64) (written int64, err error) {
if delay > 0 {
time.Sleep(time.Millisecond * time.Duration(delay))
}
return slowCopyN(sl, dst, src, n, 0)
}
func startTestServer(serverHandle func(sl *shutdownListener) http.Handler) net.Listener {
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
@@ -515,8 +696,9 @@ type shutdownListener struct {
}
func (c *shutdownListener) Close() error {
c.isShutdown = true
// Shutdown server first (waits for in-flight requests), then set isShutdown
closeErr := c.server.Shutdown(context.Background())
c.isShutdown = true
if err := ifExistAndRemove(BuildFile); err != nil {
fmt.Println(err)
}