Troubleshooting HTTP/3 QUIC Reverse Proxy for Chunked Uploads to S3 Pre-Signed URLs

package main

import (
    "context"
    "errors"
    "fmt"
    "log/slog"
    "net/http"
    "net/http/httputil"
    "net/url"
    "os"
    "os/signal"
    "time"

    "github.com/Private-repo/go-httperr"
    "github.com/Private-repo/go-reqlog"
    "github.com/quic-go/quic-go"
    "github.com/quic-go/quic-go/http3"
    "github.com/quic-go/quic-go/qlog"
)

const (
    Host                     = "0.0.0.0"
    Port                     = 4242
    webServerShutdownTimeout = 5 * time.Second
)

func main() {

    ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
    defer cancel()

    mux := http.NewServeMux()

    // V1 APIs
    mux.Handle("/post-reverse", httperr.HandlerFunc(ReverseProxy))

    hdlr := reqlog.RequestLogger(mux, nil)
    addr := fmt.Sprintf("%s:%d", Host, Port)

    server := http3.Server{
        Handler: hdlr,
        Addr:    addr,
        QUICConfig: &quic.Config{
            Tracer:    qlog.DefaultConnectionTracer,
        },
    }

    go func() {
        if err := server.ListenAndServeTLS("server.crt", "server.key"); err != nil && !errors.Is(err, http.ErrServerClosed) {
            slog.ErrorContext(ctx, "error starting server", "error", err)
            os.Exit(1)
        }
    }()

    slog.InfoContext(ctx, "started UDP-Srv", "addr", addr)

    <-ctx.Done()

    // Shutdown gracefully.
    ctx, cancel = context.WithTimeout(context.Background(), webServerShutdownTimeout)
    defer cancel()

    slog.InfoContext(ctx, "shutting down")
    if err := server.Shutdown(ctx) ; err != nil {
        slog.ErrorContext(ctx, "http server shutdown error", "error", err)
    }
}

func ReverseProxy(w http.ResponseWriter, r *http.Request) error {
    slog.Info("ReverseProxy called", "method", r.Method, "path", r.URL.Path)

    if r.Method != http.MethodPut {
        slog.Warn("Method not allowed", "method", r.Method)
        return httperr.Errorf(http.StatusMethodNotAllowed, "Method not allowed")
    }

    // Extract Pre-Signed URL
    presignedURL := r.Header.Get("X-Presigned-URL")
    if presignedURL == "" {
        slog.Warn("Missing X-Presigned-URL header")
        return httperr.Errorf(http.StatusBadRequest, "Missing X-Presigned-URL header")
    }

    // Validate Pre-Signed URL
    url, err := url.Parse(presignedURL)
    if err != nil {
        slog.Warn("Invalid X-Presigned-URL header", "error", err)
        return httperr.Errorf(http.StatusBadRequest, "Invalid X-Presigned-URL header")
    }
    slog.Info("Using Pre-Signed URL", "url", presignedURL)

    // Configure reverse proxy
    proxy := httputil.NewSingleHostReverseProxy(url)
    proxy.Director = func(req *http.Request) {
        req.URL = url
        req.Host = url.Host
        req.Method = r.Method
        req.Header = r.Header.Clone() // Clone headers
        req.Header.Del("X-Presigned-URL")
        req.ContentLength = r.ContentLength
    }

    // Handle proxy errors
    proxy.ErrorHandler = func(rw http.ResponseWriter, req *http.Request, err error) {
        slog.Error("Proxy error", "error", err)
        http.Error(rw, "Proxy error: "+err.Error(), http.StatusBadGateway)
    }

    // Serve the proxied request
    slog.Info("Forwarding request to S3")
    proxy.ServeHTTP(w, r)

    return nil
}

func (lu *Uploader) sendChunk(client *http.Client, chunkObject models.ChunkUrl, assetPath string, subtitleRelativeMap *sync.Map) error {
    var err error
    var dataReader io.Reader
    for attempt := 0; attempt < 4; attempt++ {
        if strings.Contains(assetPath, ".tar") {
            lu.logger.Info("path", "path", assetPath, "offset", chunkObject.Offset, "size", chunkObject.Size)
            path := strings.TrimSuffix(assetPath, ".tar")
            lu.logger.Info("create.tar", "path", filepath.Base(path))
            dataReader, err = lu.createTar(path, subtitleRelativeMap)
            if err != nil {
                return err
            }
            _, err = io.CopyN(io.Discard, dataReader, chunkObject.Offset)
            if err != nil {
                return err
            }

            dataReader = io.LimitReader(dataReader, chunkObject.Size)

        } else {
            var file *os.File
            file, err = os.Open(assetPath)
            if err != nil {
                return err
            }
            defer file.Close()

            dataReader = io.NewSectionReader(file, chunkObject.Offset, chunkObject.Size)

        }

        req, err := http.NewRequest(http.MethodPut, "https://localhost:4242/post-reverse", dataReader)
        if err != nil {
            return err
        }
        req.Header.Add("Content-Type", "application/octet-stream")
        req.Header.Add("X-Presigned-URL", chunkObject.UploadUrl)
        req.ContentLength = chunkObject.Size

        resp, err := client.Do(req)
        if err != nil {
            lu.logger.Error("upload.chunk.error", "attempt", attempt, "err", err)
            continue
        }

        // handle empty response
        responseBody, _ := io.ReadAll(resp.Body)
        defer resp.Body.Close()
        if resp.StatusCode != http.StatusOK {
            lu.logger.Error("upload.chunk.error", "attempt", attempt, "response", string(responseBody))
            time.Sleep(2 * time.Second)
            continue
        }
        // SUCCESS
        lu.logger.Info("upload.chunk.success", "path", assetPath, "offset", chunkObject.Offset, "size", chunkObject.Size)
        return nil
    }
    return err
}

func makeOptimizedClient() *http.Client {
    tr := &http3.Transport{
        TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
        QUICConfig:      &quic.Config{},
    }
    defer tr.Close()
    client := &http.Client{
        Transport: tr,
        Timeout:   10 * time.Minute,
    }
    return client
}