Check cached sources with upstream after the cache TTL (closes #69)
check / check (push) Canceled after 0s
check / check (push) Canceled after 0s
A cached image was served until eviction, so a change at the same URL was never picked up. After 24 hours a request checks the source with upstream, sending its stored ETag and Last-Modified. A 304 renews it with no download or conversion. New bytes replace it, and older variants are converted again. A failed check serves the cached image and puts the URL in the negative cache. Images not checked for 30 days are evicted. Variant rows keep the source's ETag and Last-Modified, so a variant whose source was evicted is checked the same way. A variant file's modification time is its source's check time when it was made or renewed, so a hit reads no database row. Model: opus-5-5
This commit is contained in:
1 parent
b3b8408d18
commit
3f487ee0ce
28 files changed
+1556
-233
No files matched your search
@@ -0,0 +1,76 @@
|
||||
package httpfetcher
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestFetch_SendsConditionalHeadersAndHandlesNotModified checks that Fetch
|
||||
// sends a non-empty ifNoneMatch and ifModifiedSince as request headers, that
|
||||
// upstream's 304 to such a request is a result with status 304 and no
|
||||
// content, and that a 304 to a request with neither is ErrUpstreamError.
|
||||
func TestFetch_SendsConditionalHeadersAndHandlesNotModified(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
etag = `"abc"`
|
||||
lastModified = "Mon, 02 Nov 2026 10:00:00 GMT"
|
||||
)
|
||||
|
||||
var (
|
||||
mu sync.Mutex
|
||||
received http.Header
|
||||
)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(
|
||||
func(w http.ResponseWriter, r *http.Request) {
|
||||
mu.Lock()
|
||||
received = r.Header.Clone()
|
||||
mu.Unlock()
|
||||
|
||||
w.WriteHeader(http.StatusNotModified)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
f, _ := newServerFetcher(t, srv, nil)
|
||||
|
||||
res, err := f.Fetch(testContext(t), upstreamSource("/image"), etag, lastModified)
|
||||
if err != nil {
|
||||
t.Fatalf("conditional Fetch() error = %v", err)
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(res.Content)
|
||||
_ = res.Content.Close()
|
||||
|
||||
mu.Lock()
|
||||
t.Logf("status = %d, body = %q, request headers = %v", res.StatusCode, body, received)
|
||||
gotIfNoneMatch := received.Get("If-None-Match")
|
||||
gotIfModifiedSince := received.Get("If-Modified-Since")
|
||||
mu.Unlock()
|
||||
|
||||
if err != nil || len(body) != 0 {
|
||||
t.Errorf("304 content = %q, error = %v; want empty", body, err)
|
||||
}
|
||||
|
||||
if res.StatusCode != http.StatusNotModified {
|
||||
t.Errorf("StatusCode = %d, want 304", res.StatusCode)
|
||||
}
|
||||
|
||||
if gotIfNoneMatch != etag || gotIfModifiedSince != lastModified {
|
||||
t.Errorf("If-None-Match = %q, If-Modified-Since = %q; want %q and %q",
|
||||
gotIfNoneMatch, gotIfModifiedSince, etag, lastModified)
|
||||
}
|
||||
|
||||
if held := len(f.allHostsSemaphore); held != 0 {
|
||||
t.Errorf("connection still held after the 304: %d held", held)
|
||||
}
|
||||
|
||||
_, err = f.Fetch(testContext(t), upstreamSource("/image"), "", "")
|
||||
if !errors.Is(err, ErrUpstreamError) {
|
||||
t.Errorf("unconditional Fetch() answered 304: error = %v, want ErrUpstreamError", err)
|
||||
}
|
||||
}
|
||||
@@ -51,12 +51,12 @@ func TestDialContextReplacesOnlyTheDialer(t *testing.T) {
|
||||
t.Errorf("body = %q, want %q", body, imagePayload)
|
||||
}
|
||||
|
||||
_, err := f.Fetch(testContext(t), imageSourceOn("127.0.0.1"))
|
||||
_, err := f.Fetch(testContext(t), imageSourceOn("127.0.0.1"), "", "")
|
||||
if !errors.Is(err, ErrSSRFBlocked) {
|
||||
t.Errorf("Fetch(loopback URL) error = %v, want ErrSSRFBlocked", err)
|
||||
}
|
||||
|
||||
_, err = f.Fetch(testContext(t), upstreamSource("/redirect/private"))
|
||||
_, err = f.Fetch(testContext(t), upstreamSource("/redirect/private"), "", "")
|
||||
if !errors.Is(err, ErrSSRFBlocked) {
|
||||
t.Errorf("Fetch(/redirect/private) error = %v, want ErrSSRFBlocked", err)
|
||||
}
|
||||
|
||||
@@ -160,7 +160,7 @@ func testContext(t *testing.T) context.Context {
|
||||
func fetchImage(t *testing.T, f *HTTPFetcher, path string) *FetchResult {
|
||||
t.Helper()
|
||||
|
||||
res, err := f.Fetch(testContext(t), upstreamSource(path))
|
||||
res, err := f.Fetch(testContext(t), upstreamSource(path), "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("Fetch(%s) error = %v", path, err)
|
||||
}
|
||||
@@ -172,7 +172,7 @@ func fetchImage(t *testing.T, f *HTTPFetcher, path string) *FetchResult {
|
||||
func fetchExpectError(t *testing.T, f *HTTPFetcher, path string) error {
|
||||
t.Helper()
|
||||
|
||||
res, err := f.Fetch(testContext(t), upstreamSource(path))
|
||||
res, err := f.Fetch(testContext(t), upstreamSource(path), "", "")
|
||||
if err == nil {
|
||||
_ = res.Content.Close()
|
||||
|
||||
@@ -224,7 +224,7 @@ func TestFetchRedirectToPrivateIPBlocked(t *testing.T) {
|
||||
srv := startUpstream(t)
|
||||
f, dialer := newServerFetcher(t, srv, nil)
|
||||
|
||||
_, err := f.Fetch(testContext(t), upstreamSource("/redirect/private"))
|
||||
_, err := f.Fetch(testContext(t), upstreamSource("/redirect/private"), "", "")
|
||||
if !errors.Is(err, ErrSSRFBlocked) {
|
||||
t.Fatalf("Fetch() error = %v, want ErrSSRFBlocked", err)
|
||||
}
|
||||
@@ -624,7 +624,7 @@ func TestFetchSchemeFollowsAllowHTTP(t *testing.T) {
|
||||
DialContext: dialer.dialContext,
|
||||
})
|
||||
|
||||
res, err := f.Fetch(testContext(t), upstreamSource("/image"))
|
||||
res, err := f.Fetch(testContext(t), upstreamSource("/image"), "", "")
|
||||
if err == nil {
|
||||
_ = res.Content.Close()
|
||||
}
|
||||
|
||||
@@ -80,8 +80,13 @@ var (
|
||||
|
||||
// Fetcher retrieves content from upstream origins.
|
||||
type Fetcher interface {
|
||||
// Fetch retrieves the content of the source image.
|
||||
Fetch(ctx context.Context, source imagerequest.Source) (*FetchResult, error)
|
||||
// Fetch retrieves the content of the source image. A non-empty
|
||||
// ifNoneMatch or ifModifiedSince is sent as the request header of that
|
||||
// name, and upstream may then answer 304 Not Modified with no content.
|
||||
Fetch(
|
||||
ctx context.Context, source imagerequest.Source,
|
||||
ifNoneMatch, ifModifiedSince string,
|
||||
) (*FetchResult, error)
|
||||
}
|
||||
|
||||
// FetchResult contains the result of fetching from upstream.
|
||||
@@ -223,9 +228,13 @@ func New(config *Config) *HTTPFetcher {
|
||||
// Fetch retrieves the source image with SSRF protection, over https, or over
|
||||
// http when AllowHTTP is set. When MaxConnections fetches are already in
|
||||
// progress, it waits up to ConnectionWaitTimeout for one to finish, then
|
||||
// fails with ErrTooManyConnections.
|
||||
// fails with ErrTooManyConnections. A non-empty ifNoneMatch or
|
||||
// ifModifiedSince is sent as the If-None-Match or If-Modified-Since header.
|
||||
// Upstream's 304 Not Modified to such a request is a result with StatusCode
|
||||
// 304 and empty Content; to any other request it is ErrUpstreamError.
|
||||
func (f *HTTPFetcher) Fetch(
|
||||
ctx context.Context, source imagerequest.Source,
|
||||
ifNoneMatch, ifModifiedSince string,
|
||||
) (*FetchResult, error) {
|
||||
sourceURL := source.URL(f.config.AllowHTTP)
|
||||
|
||||
@@ -252,17 +261,7 @@ func (f *HTTPFetcher) Fetch(
|
||||
req := &http.Request{
|
||||
Method: http.MethodGet,
|
||||
URL: sourceURL,
|
||||
Header: make(http.Header),
|
||||
}
|
||||
|
||||
req.Header.Set("User-Agent", userAgent)
|
||||
req.Header.Set("Accept", strings.Join(allowedContentTypes(), ", "))
|
||||
|
||||
// The ID of the request this fetch serves, so the fetch can be found in
|
||||
// the upstream host's logs
|
||||
requestID := middleware.GetReqID(ctx)
|
||||
if requestID != "" {
|
||||
req.Header.Set(middleware.RequestIDHeader, requestID)
|
||||
Header: requestHeaders(ctx, ifNoneMatch, ifModifiedSince),
|
||||
}
|
||||
|
||||
// Use httptrace to capture connection details
|
||||
@@ -291,6 +290,20 @@ func (f *HTTPFetcher) Fetch(
|
||||
return nil, fmt.Errorf("upstream request failed: %w", err)
|
||||
}
|
||||
|
||||
// The deferred release gives the connection back, as no body follows
|
||||
conditional := ifNoneMatch != "" || ifModifiedSince != ""
|
||||
if conditional && resp.StatusCode == http.StatusNotModified {
|
||||
_ = resp.Body.Close()
|
||||
|
||||
return &FetchResult{
|
||||
Content: http.NoBody,
|
||||
Headers: resp.Header,
|
||||
StatusCode: resp.StatusCode,
|
||||
FetchDurationMs: fetchDuration.Milliseconds(),
|
||||
RemoteAddr: remoteAddr,
|
||||
}, nil
|
||||
}
|
||||
|
||||
result, err := f.buildResult(resp, remoteAddr, fetchDuration, release)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -303,6 +316,33 @@ func (f *HTTPFetcher) Fetch(
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// requestHeaders returns the headers of a fetch: User-Agent, Accept, the
|
||||
// ID of the request the fetch serves, so the fetch can be found in the
|
||||
// upstream host's logs, and a non-empty ifNoneMatch or ifModifiedSince.
|
||||
func requestHeaders(
|
||||
ctx context.Context, ifNoneMatch, ifModifiedSince string,
|
||||
) http.Header {
|
||||
header := make(http.Header)
|
||||
|
||||
header.Set("User-Agent", userAgent)
|
||||
header.Set("Accept", strings.Join(allowedContentTypes(), ", "))
|
||||
|
||||
requestID := middleware.GetReqID(ctx)
|
||||
if requestID != "" {
|
||||
header.Set(middleware.RequestIDHeader, requestID)
|
||||
}
|
||||
|
||||
if ifNoneMatch != "" {
|
||||
header.Set("If-None-Match", ifNoneMatch)
|
||||
}
|
||||
|
||||
if ifModifiedSince != "" {
|
||||
header.Set("If-Modified-Since", ifModifiedSince)
|
||||
}
|
||||
|
||||
return header
|
||||
}
|
||||
|
||||
// acquireConnection takes a slot for host, then one of the slots shared by
|
||||
// all hosts, and returns the func that gives both back. The host's slot
|
||||
// comes first, so fetches queued for one busy host hold no shared slot.
|
||||
|
||||
@@ -210,7 +210,7 @@ func TestMockFetcher_FetchesFile(t *testing.T) {
|
||||
m := NewMock(mockFS)
|
||||
|
||||
result, err := m.Fetch(context.Background(),
|
||||
imagerequest.Source{Host: testHost, Path: "/images/photo.jpg"})
|
||||
imagerequest.Source{Host: testHost, Path: "/images/photo.jpg"}, "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("Fetch() error = %v", err)
|
||||
}
|
||||
@@ -242,7 +242,7 @@ func TestMockFetcher_MissingFileReturnsUpstreamError(t *testing.T) {
|
||||
m := NewMock(mockFS)
|
||||
|
||||
_, err := m.Fetch(context.Background(),
|
||||
imagerequest.Source{Host: testHost, Path: "/missing.jpg"})
|
||||
imagerequest.Source{Host: testHost, Path: "/missing.jpg"}, "", "")
|
||||
if !errors.Is(err, ErrUpstreamError) {
|
||||
t.Errorf("Fetch() error = %v, want ErrUpstreamError", err)
|
||||
}
|
||||
@@ -259,7 +259,7 @@ func TestMockFetcher_RespectsContextCancellation(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
|
||||
_, err := m.Fetch(ctx, imagerequest.Source{Host: testHost, Path: "/photo.jpg"})
|
||||
_, err := m.Fetch(ctx, imagerequest.Source{Host: testHost, Path: "/photo.jpg"}, "", "")
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("Fetch() error = %v, want context.Canceled", err)
|
||||
}
|
||||
|
||||
@@ -45,12 +45,12 @@ func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) {
|
||||
f, _ := newServerFetcher(t, srv, cfg)
|
||||
f.connectionWaitTimeout = 100 * time.Millisecond
|
||||
|
||||
first, err := f.Fetch(testContext(t), imageSourceOnPort(81))
|
||||
first, err := f.Fetch(testContext(t), imageSourceOnPort(81), "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("first Fetch() error = %v", err)
|
||||
}
|
||||
|
||||
second, err := f.Fetch(testContext(t), imageSourceOnPort(82))
|
||||
second, err := f.Fetch(testContext(t), imageSourceOnPort(82), "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("second Fetch() error = %v", err)
|
||||
}
|
||||
@@ -59,7 +59,7 @@ func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) {
|
||||
|
||||
start := time.Now()
|
||||
|
||||
_, err = f.Fetch(testContext(t), imageSourceOnPort(83))
|
||||
_, err = f.Fetch(testContext(t), imageSourceOnPort(83), "", "")
|
||||
if !errors.Is(err, ErrTooManyConnections) {
|
||||
t.Fatalf("third Fetch() error = %v, want ErrTooManyConnections", err)
|
||||
}
|
||||
@@ -78,7 +78,7 @@ func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) {
|
||||
t.Fatalf("close first body: %v", err)
|
||||
}
|
||||
|
||||
third, err := f.Fetch(testContext(t), imageSourceOnPort(83))
|
||||
third, err := f.Fetch(testContext(t), imageSourceOnPort(83), "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("Fetch() after a response was closed: error = %v", err)
|
||||
}
|
||||
@@ -105,7 +105,7 @@ func TestFetchFreesHostSlotWhenContextEndsWaitingForConnection(t *testing.T) {
|
||||
|
||||
f, _ := newServerFetcher(t, srv, cfg)
|
||||
|
||||
first, err := f.Fetch(testContext(t), imageSourceOnPort(81))
|
||||
first, err := f.Fetch(testContext(t), imageSourceOnPort(81), "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("first Fetch() error = %v", err)
|
||||
}
|
||||
@@ -115,7 +115,7 @@ func TestFetchFreesHostSlotWhenContextEndsWaitingForConnection(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
_, err = f.Fetch(ctx, imageSourceOnPort(82))
|
||||
_, err = f.Fetch(ctx, imageSourceOnPort(82), "", "")
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("second Fetch() error = %v, want context.DeadlineExceeded", err)
|
||||
}
|
||||
@@ -140,7 +140,7 @@ func TestFetchRemovesIdleHostSemaphores(t *testing.T) {
|
||||
|
||||
for i := range 100 {
|
||||
wg.Go(func() {
|
||||
res, err := f.Fetch(ctx, imageSourceOnPort(1+i%50))
|
||||
res, err := f.Fetch(ctx, imageSourceOnPort(1+i%50), "", "")
|
||||
if err != nil {
|
||||
t.Errorf("Fetch() error = %v", err)
|
||||
|
||||
@@ -179,12 +179,12 @@ func TestFetchRemovesHostSemaphoreWhenNoConnection(t *testing.T) {
|
||||
f, _ := newServerFetcher(t, srv, cfg)
|
||||
f.connectionWaitTimeout = 100 * time.Millisecond
|
||||
|
||||
open, err := f.Fetch(testContext(t), imageSourceOnPort(81))
|
||||
open, err := f.Fetch(testContext(t), imageSourceOnPort(81), "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("first Fetch() error = %v", err)
|
||||
}
|
||||
|
||||
_, err = f.Fetch(testContext(t), imageSourceOnPort(82))
|
||||
_, err = f.Fetch(testContext(t), imageSourceOnPort(82), "", "")
|
||||
if !errors.Is(err, ErrTooManyConnections) {
|
||||
t.Fatalf("Fetch() from another host: error = %v, "+
|
||||
"want ErrTooManyConnections", err)
|
||||
@@ -193,7 +193,7 @@ func TestFetchRemovesHostSemaphoreWhenNoConnection(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
_, err = f.Fetch(ctx, imageSourceOnPort(81))
|
||||
_, err = f.Fetch(ctx, imageSourceOnPort(81), "", "")
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("Fetch() from the busy host: error = %v, "+
|
||||
"want context.DeadlineExceeded", err)
|
||||
@@ -205,7 +205,7 @@ func TestFetchRemovesHostSemaphoreWhenNoConnection(t *testing.T) {
|
||||
ctx, cancel = context.WithTimeout(t.Context(), 100*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
_, err = f.Fetch(ctx, imageSourceOnPort(83))
|
||||
_, err = f.Fetch(ctx, imageSourceOnPort(83), "", "")
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("Fetch() from a host with nothing open: error = %v, "+
|
||||
"want context.DeadlineExceeded", err)
|
||||
@@ -255,7 +255,7 @@ func TestFetchReleasesConnectionOnError(t *testing.T) {
|
||||
f, _ := newServerFetcher(t, srv, cfg)
|
||||
f.connectionWaitTimeout = 100 * time.Millisecond
|
||||
|
||||
_, err := f.Fetch(testContext(t), tc.source)
|
||||
_, err := f.Fetch(testContext(t), tc.source, "", "")
|
||||
if !errors.Is(err, tc.want) {
|
||||
t.Fatalf("Fetch() error = %v, want %v", err, tc.want)
|
||||
}
|
||||
|
||||
@@ -24,9 +24,10 @@ func NewMock(fsys fs.FS) *MockFetcher {
|
||||
return &MockFetcher{fs: fsys}
|
||||
}
|
||||
|
||||
// Fetch retrieves the source's content from the mock filesystem.
|
||||
// Fetch retrieves the source's content from the mock filesystem. It ignores
|
||||
// ifNoneMatch and ifModifiedSince, and always sends the content.
|
||||
func (m *MockFetcher) Fetch(
|
||||
ctx context.Context, source imagerequest.Source,
|
||||
ctx context.Context, source imagerequest.Source, _, _ string,
|
||||
) (*FetchResult, error) {
|
||||
// Check context cancellation
|
||||
select {
|
||||
|
||||
@@ -35,7 +35,7 @@ func TestFetchSendsRequestID(t *testing.T) {
|
||||
|
||||
ctx := context.WithValue(testContext(t), middleware.RequestIDKey, requestID)
|
||||
|
||||
res, err := f.Fetch(ctx, upstreamSource("/image"))
|
||||
res, err := f.Fetch(ctx, upstreamSource("/image"), "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("Fetch() error = %v", err)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user