test and fix bugs found by cline

memory-bank/activeContext.md

@@ -1,22 +1,24 @@
 # 当前工作重点
 
-当前的主要任务是熟悉并理解现有的 `cache-proxy` 项目。这包括分析其代码结构、核心算法和设计模式。
+当前的工作重点是根据 `progress.md` 中的待办事项，开始对项目进行优化和功能增强。在完成了全面的测试覆盖后，我们对现有代码的稳定性和正确性有了很强的信心，可以安全地进行重构。
 
 ## 近期变更
 
--   **初始化 Memory Bank**: 创建了 Memory Bank 的核心文档，包括：
+-   **完成 `server_test.go`**:
-    -   `projectbrief.md`
+    -   补全了 `server_test.go` 中所有待办的测试用例，包括对 `X-Accel-Redirect` 和路径穿越攻击的测试。
-    -   `productContext.md`
+    -   对所有测试用例的注释进行了审查和修正，确保注释与代码的实际行为保持一致。
-    -   `systemPatterns.md`
+    -   所有测试均已通过，为后续的开发和重构工作奠定了坚实的基础。
-    -   `techContext.md`
+-   **更新 `progress.md`**:
-    -   `activeContext.md`
+    -   将“增加更全面的单元测试和集成测试”标记为已完成。
-    -   `progress.md`
 
 ## 后续步骤
 
-1.  **验证理解**: 与项目所有者沟通，确认对项目的设计和功能的理解是否准确。
+1.  **代码重构**:
-2.  **确定开发方向**: 明确项目当前是否存在需要修复的 bug、需要优化的性能瓶颈或需要开发的新功能。
+    -   根据 `progress.md` 的待办事项，首先考虑对 `server.go` 中的复杂函数（如 `streamOnline`）进行重构，以提高代码的可读性和可维护性。
-3.  **完善文档**: 根据后续的开发工作，持续更新和完善 Memory Bank 中的文档。
+2.  **功能增强**:
+    -   在代码结构优化后，可以开始考虑实现新的功能，例如增加对 S3 等新存储后端的支持，或实现更复杂的负载均衡策略。
+3.  **持续文档更新**:
+    -   在进行重构或添加新功能时，同步更新 `systemPatterns.md` 和其他相关文档，以记录新的设计决策。
 
 ## 重要模式与偏好
 

memory-bank/progress.md

@@ -20,6 +20,11 @@
     -   能够正确处理对同一文件的多个并发请求，确保只下载一次。
 -   **加速下载**:
     -   支持通过 `X-Sendfile` / `X-Accel-Redirect` 头将文件发送委托给前端服务器（如 Nginx）。
+-   **全面的测试覆盖**:
+    -   完成了 `server_test.go` 的实现，为所有核心功能提供了单元测试和集成测试。
+    -   测试覆盖了正常流程、边缘情况（如超时、上游失败）和安全（如路径穿越）等方面。
+    -   对测试代码和注释进行了审查，确保其准确性和一致性。
+    -   所有测试均已通过，验证了现有代码的健壮性。
 
 ## 待办事项
 
@@ -29,7 +34,6 @@
     -   增加更详细的监控和指标（如 Prometheus metrics）。
 -   **代码优化**:
     -   对 `server.go` 中的一些复杂函数（如 `streamOnline`）进行重构，以提高可读性和可维护性。
-    -   增加更全面的单元测试和集成测试。
 
 ## 已知问题
 

server.go

@@ -25,12 +25,17 @@ const (
 
 var zeroTime time.Time
 
+var preclosedChan = make(chan struct{})
+
+func init() {
+	close(preclosedChan)
+}
+
 var (
 	httpClient = http.Client{
 		// check allowed redirect
 		CheckRedirect: func(req *http.Request, via []*http.Request) error {
-			lastRequest := via[len(via)-1]
+			if allowedRedirect, ok := req.Context().Value(reqCtxAllowedRedirect).(string); ok {
-			if allowedRedirect, ok := lastRequest.Context().Value(reqCtxAllowedRedirect).(string); ok {
 				if matched, err := regexp.MatchString(allowedRedirect, req.URL.String()); err != nil {
 					return err
 				} else if !matched {
@@ -147,7 +152,7 @@ func (server *Server) HandleRequestWithCache(w http.ResponseWriter, r *http.Requ
 	} else if localStatus != localNotExists {
 		if localStatus == localExistsButNeedHead {
 			if ranged {
-				server.streamOnline(nil, r, mtime, fullpath)
+				<-server.streamOnline(nil, r, mtime, fullpath)
 				server.serveFile(w, r, fullpath)
 			} else {
 				server.streamOnline(w, r, mtime, fullpath)
@@ -157,7 +162,7 @@ func (server *Server) HandleRequestWithCache(w http.ResponseWriter, r *http.Requ
 		}
 	} else {
 		if ranged {
-			server.streamOnline(nil, r, mtime, fullpath)
+			<-server.streamOnline(nil, r, mtime, fullpath)
 			server.serveFile(w, r, fullpath)
 		} else {
 			server.streamOnline(w, r, mtime, fullpath)
@@ -209,7 +214,7 @@ func (server *Server) checkLocal(w http.ResponseWriter, _ *http.Request, key str
 	return localNotExists, zeroTime, nil
 }
 
-func (server *Server) streamOnline(w http.ResponseWriter, r *http.Request, mtime time.Time, key string) {
+func (server *Server) streamOnline(w http.ResponseWriter, r *http.Request, mtime time.Time, key string) <-chan struct{} {
 	memoryObject, exists := server.o[r.URL.Path]
 	locked := false
 	defer func() {
@@ -241,32 +246,39 @@ func (server *Server) streamOnline(w http.ResponseWriter, r *http.Request, mtime
 				slog.With("error", err).Warn("failed to stream response with existing memory object")
 			}
 		}
+		return preclosedChan
 	} else {
 		slog.With("mtime", mtime).Debug("checking fastest upstream")
-		selectedIdx, response, chunks, err := server.fastesUpstream(r, mtime)
+		selectedIdx, response, chunks, err := server.fastestUpstream(r, mtime)
 		if chunks == nil && mtime != zeroTime {
 			slog.With("upstreamIdx", selectedIdx, "key", key).Debug("not modified. using local version")
 			if w != nil {
 				server.serveFile(w, r, key)
 			}
-			return
+			return preclosedChan
 		}
 
 		if err != nil {
 			slog.With("error", err).Warn("failed to select fastest upstream")
-			http.Error(w, err.Error(), http.StatusInternalServerError)
+			if w != nil {
-			return
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+			}
+			return preclosedChan
 		}
 		if selectedIdx == -1 || response == nil || chunks == nil {
 			slog.Debug("no upstream is selected")
-			http.NotFound(w, r)
+			if w != nil {
-			return
+				http.NotFound(w, r)
+			}
+			return preclosedChan
 		}
 		if response.StatusCode == http.StatusNotModified {
 			slog.With("upstreamIdx", selectedIdx).Debug("not modified. using local version")
 			os.Chtimes(key, zeroTime, time.Now())
-			server.serveFile(w, r, key)
+			if w != nil {
-			return
+				server.serveFile(w, r, key)
+			}
+			return preclosedChan
 		}
 
 		slog.With(
@@ -337,26 +349,28 @@ func (server *Server) streamOnline(w http.ResponseWriter, r *http.Request, mtime
 			hijacker, ok := w.(http.Hijacker)
 			if !ok {
 				logger.Warn("response writer is not a hijacker. failed to set lingering")
-				return
+				return preclosedChan
 			}
 			conn, _, err := hijacker.Hijack()
 			if err != nil {
 				logger.With("error", err).Warn("hijack failed. failed to set lingering")
-				return
+				return preclosedChan
 			}
 			defer conn.Close()
 
 			tcpConn, ok := conn.(*net.TCPConn)
 			if !ok {
 				logger.With("error", err).Warn("connection is not a *net.TCPConn. failed to set lingering")
-				return
+				return preclosedChan
 			}
 			if err := tcpConn.SetLinger(0); err != nil {
 				logger.With("error", err).Warn("failed to set lingering")
-				return
+				return preclosedChan
 			}
 			logger.Debug("connection set to linger. it will be reset once the conn.Close is called")
 		}
+
+		fileWrittenCh := make(chan struct{})
 		go func() {
 			defer func() {
 				server.lu.Lock()
@@ -364,6 +378,7 @@ func (server *Server) streamOnline(w http.ResponseWriter, r *http.Request, mtime
 
 				delete(server.o, r.URL.Path)
 				slog.Debug("memory object released")
+				close(fileWrittenCh)
 			}()
 
 			if err == nil {
@@ -428,10 +443,11 @@ func (server *Server) streamOnline(w http.ResponseWriter, r *http.Request, mtime
 			}
 		}()
 
+		return fileWrittenCh
 	}
 }
 
-func (server *Server) fastesUpstream(r *http.Request, lastModified time.Time) (resultIdx int, resultResponse *http.Response, resultCh chan Chunk, resultErr error) {
+func (server *Server) fastestUpstream(r *http.Request, lastModified time.Time) (resultIdx int, resultResponse *http.Response, resultCh chan Chunk, resultErr error) {
 	returnLock := &sync.Mutex{}
 	upstreams := len(server.Upstreams)
 	cancelFuncs := make([]func(), upstreams)
@@ -619,7 +635,7 @@ func (server *Server) tryUpstream(ctx context.Context, upstreamIdx, priority int
 	}
 	streaming := false
 	defer func() {
-		if !streaming {
+		if !streaming && response != nil {
 			response.Body.Close()
 		}
 	}()
@@ -684,6 +700,7 @@ func (server *Server) tryUpstream(ctx context.Context, upstreamIdx, priority int
 	streaming = true
 	go func() {
 		defer close(ch)
+		defer response.Body.Close()
 
 		for {
 			buffer := make([]byte, server.Misc.ChunkBytes)