feat(registry): 开启 manifest 删除，GC sidecar 在 gate 只读窗口内回收 blob

internal/imagepush/push.go

@@ -25,6 +25,7 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -57,6 +58,13 @@ type Pusher struct {
 	Log io.Writer
 	// Attempts bounds retries of one blob upload or the manifest PUT. Zero means 3.
 	Attempts int
+	// MaxWait bounds the total time a push waits out 503 answers that carry
+	// Retry-After — the gate's read-only window while garbage collection runs.
+	// Those waits do not use up Attempts. Zero means 20 minutes.
+	MaxWait time.Duration
+
+	// after is time.After, replaced in tests.
+	after func(time.Duration) <-chan time.Time
 }
 
 // Ref is a parsed host/repository:tag reference.
@@ -296,33 +304,62 @@ func (p *Pusher) do(ctx context.Context, method, target string, body io.Reader,
 }
 
 // retry runs fn up to Attempts times, backing off between tries. A refusal the
-// registry will repeat (401/403/4xx other than 408/429) is returned at once.
+// registry will repeat (401/403/4xx other than 408/429) is returned at once. A 503
+// with Retry-After is waited out without using up an attempt, for as long as
+// MaxWait allows.
 func (p *Pusher) retry(ctx context.Context, fn func() error) error {
 	attempts := p.Attempts
 	if attempts <= 0 {
 		attempts = 3
 	}
-	var err error
-	for i := 0; i < attempts; i++ {
+	maxWait := p.MaxWait
+	if maxWait <= 0 {
+		maxWait = 20 * time.Minute
+	}
+	var (
+		err    error
+		waited time.Duration
+	)
+	for i := 0; i < attempts; {
 		if err = fn(); err == nil {
 			return nil
 		}
 		var se *StatusError
+		if errors.As(err, &se) && se.Code == http.StatusServiceUnavailable && se.RetryAfter > 0 && waited+se.RetryAfter <= maxWait {
+			p.logf("registry unavailable, waiting %s: %s", se.RetryAfter, se.Body)
+			if err := p.sleep(ctx, se.RetryAfter); err != nil {
+				return err
+			}
+			waited += se.RetryAfter
+			continue
+		}
 		if errors.As(err, &se) && se.Code >= 400 && se.Code < 500 && se.Code != http.StatusRequestTimeout && se.Code != http.StatusTooManyRequests {
 			return err
 		}
-		if i+1 < attempts {
+		i++
+		if i < attempts {
 			p.logf("retrying after: %v", err)
-			select {
-			case <-ctx.Done():
-				return ctx.Err()
-			case <-time.After(time.Duration(i+1) * time.Second):
+			if err := p.sleep(ctx, time.Duration(i)*time.Second); err != nil {
+				return err
 			}
 		}
 	}
 	return err
 }
 
+func (p *Pusher) sleep(ctx context.Context, d time.Duration) error {
+	after := p.after
+	if after == nil {
+		after = time.After
+	}
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-after(d):
+		return nil
+	}
+}
+
 func (p *Pusher) logf(format string, args ...any) {
 	if p.Log != nil {
 		fmt.Fprintf(p.Log, format+"\n", args...)
@@ -334,6 +371,8 @@ type StatusError struct {
 	Op   string
 	Code int
 	Body string
+	// RetryAfter is the registry's Retry-After, when it sent one in seconds.
+	RetryAfter time.Duration
 }
 
 func (e *StatusError) Error() string {
@@ -342,7 +381,11 @@ func (e *StatusError) Error() string {
 
 func statusError(op string, resp *http.Response) error {
 	b, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))
-	return &StatusError{Op: op, Code: resp.StatusCode, Body: strings.TrimSpace(string(b))}
+	se := &StatusError{Op: op, Code: resp.StatusCode, Body: strings.TrimSpace(string(b))}
+	if sec, err := strconv.Atoi(resp.Header.Get("Retry-After")); err == nil && sec > 0 {
+		se.RetryAfter = time.Duration(sec) * time.Second
+	}
+	return se
 }
 
 // openEntry returns a reader positioned at the named tar entry. The caller closes

internal/imagepush/push_test.go

@@ -19,6 +19,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"felis.lolicon.best/internal/registrygate"
 )
@@ -191,6 +192,83 @@ func TestPushThroughTheGate(t *testing.T) {
 	}
 }
 
+// TestPushWaitsOutTheGCWindow: while the gate holds the registry read-only for
+// garbage collection, a push waits on Retry-After (without spending its attempts)
+// and completes once the window closes; past MaxWait it gives up with the 503.
+func TestPushWaitsOutTheGCWindow(t *testing.T) {
+	reg := newFakeRegistry()
+	upstream := httptest.NewServer(reg)
+	t.Cleanup(upstream.Close)
+	u, _ := url.Parse(upstream.URL)
+	g := registrygate.New(u, map[string]string{registrygate.PrincipalBuild: "build-secret"}, nil)
+	g.SetQuiet(0)
+	gate := httptest.NewServer(g)
+	t.Cleanup(gate.Close)
+	maint := httptest.NewServer(g.MaintHandler())
+	t.Cleanup(maint.Close)
+	host := strings.TrimPrefix(gate.URL, "http://")
+	setReadOnly := func(on bool) {
+		t.Helper()
+		path := "/readwrite"
+		if on {
+			path = "/readonly?lease=600"
+		}
+		resp, err := http.Post(maint.URL+path, "text/plain", nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		resp.Body.Close()
+		if resp.StatusCode != http.StatusOK {
+			t.Fatalf("POST %s = %d", path, resp.StatusCode)
+		}
+	}
+
+	ref := host + "/user-uploads/sub-gc:latest"
+	tarPath := writeTarball(t, ref, []byte("layer"))
+	setReadOnly(true)
+	var waits []time.Duration
+	p := &Pusher{Scheme: "http", Username: registrygate.PrincipalBuild, Password: "build-secret", Attempts: 1}
+	p.after = func(d time.Duration) <-chan time.Time {
+		waits = append(waits, d)
+		if len(waits) == 3 {
+			setReadOnly(false) // the sweep finished
+		}
+		ch := make(chan time.Time, 1)
+		ch <- time.Time{}
+		return ch
+	}
+	if _, err := p.Push(context.Background(), tarPath, ref); err != nil {
+		t.Fatalf("push across the window: %v", err)
+	}
+	if len(waits) != 3 || waits[0] != 30*time.Second {
+		t.Fatalf("waits = %v, want three Retry-After (30s) waits", waits)
+	}
+	if reg.manifests["user-uploads/sub-gc:latest"] == nil {
+		t.Fatal("no manifest after the window closed")
+	}
+
+	// A window that outlasts MaxWait fails the push with the 503.
+	setReadOnly(true)
+	ref2 := host + "/user-uploads/sub-gc2:latest"
+	tar2 := writeTarball(t, ref2, []byte("other layer"))
+	waits = nil
+	p.MaxWait = time.Minute
+	p.after = func(d time.Duration) <-chan time.Time {
+		waits = append(waits, d)
+		ch := make(chan time.Time, 1)
+		ch <- time.Time{}
+		return ch
+	}
+	_, err := p.Push(context.Background(), tar2, ref2)
+	var se *StatusError
+	if !errors.As(err, &se) || se.Code != http.StatusServiceUnavailable {
+		t.Fatalf("push past MaxWait = %v, want the 503", err)
+	}
+	if len(waits) != 2 {
+		t.Fatalf("waits = %v, want 2 (60s of MaxWait at 30s each)", waits)
+	}
+}
+
 func TestPushIntoAReservedRepoIsRefusedWithoutRetry(t *testing.T) {
 	reg, host := startStack(t)
 	ref := host + "/felis/felis:v0.1.0"