detached a235943

feat: implement chunked blob upload for native registry push

commit
a235943db4d7de605d3dc4cef6b761805bda4827
author
brice · · 2026-03-14 13:56 UTC
parent
f8212fb
signature
unsigned
2 files · +427 -0
Unified
UnifiedSplit
M pkg/registrypush/push.go +218 -0
@@ -2,8 +2,10 @@ package registrypush
22
33 import (
44 "context"
5+ "fmt"
56 "io"
67 "net/http"
8+ "net/url"
79
810 v1 "github.com/google/go-containerregistry/pkg/v1"
911 )
@@ -72,8 +74,224 @@ func (p *Pusher) Push(ctx context.Context, ref string) error {
7274 return nil
7375 }
7476
77+// UploadBlob uploads a single blob to the registry using chunked uploads.
78+// If the blob already exists, it is skipped.
79+func (p *Pusher) UploadBlob(ctx context.Context, baseURL, repo, digest string, size int64, content io.Reader, auth string) error {
80+ if err := p.checkBlobExists(ctx, baseURL, repo, digest, auth); err == nil {
81+ return nil
82+ } else if err != errBlobMissing {
83+ return err
84+ }
85+
86+ uploadURL, err := p.startBlobUpload(ctx, baseURL, repo, auth)
87+ if err != nil {
88+ return err
89+ }
90+
91+ chunks := chunkCount(size, p.chunkSize)
92+ if p.progress != nil {
93+ if _, err := fmt.Fprintf(p.progress, "Uploading blob %s (%d bytes, %d chunks)\n", digest, size, chunks); err != nil {
94+ return fmt.Errorf("failed to write upload progress: %w", err)
95+ }
96+ }
97+
98+ uploadURL, err = p.uploadBlobChunks(ctx, uploadURL, content, auth)
99+ if err != nil {
100+ return err
101+ }
102+
103+ return p.finalizeBlobUpload(ctx, uploadURL, digest, auth)
104+}
105+
75106 // defaultImageSource reads an image from the local Docker/Podman daemon.
76107 func defaultImageSource(ctx context.Context, ref string) (v1.Image, error) {
77108 // Implementation in Task 3
78109 return nil, nil
79110 }
111+
112+var errBlobMissing = fmt.Errorf("blob missing")
113+
114+func (p *Pusher) checkBlobExists(ctx context.Context, baseURL, repo, digest, auth string) error {
115+ req, err := http.NewRequestWithContext(ctx, http.MethodHead, fmt.Sprintf("%s/v2/%s/blobs/%s", baseURL, repo, digest), nil)
116+ if err != nil {
117+ return fmt.Errorf("failed to create blob HEAD request: %w", err)
118+ }
119+ setAuthHeader(req, auth)
120+
121+ resp, err := p.httpClient().Do(req)
122+ if err != nil {
123+ return fmt.Errorf("failed to check blob existence: %w", err)
124+ }
125+ defer resp.Body.Close()
126+
127+ switch resp.StatusCode {
128+ case http.StatusOK:
129+ return nil
130+ case http.StatusNotFound:
131+ return errBlobMissing
132+ default:
133+ return fmt.Errorf("blob existence check returned status %d", resp.StatusCode)
134+ }
135+}
136+
137+func (p *Pusher) startBlobUpload(ctx context.Context, baseURL, repo, auth string) (string, error) {
138+ req, err := http.NewRequestWithContext(ctx, http.MethodPost, fmt.Sprintf("%s/v2/%s/blobs/uploads/", baseURL, repo), nil)
139+ if err != nil {
140+ return "", fmt.Errorf("failed to create blob upload start request: %w", err)
141+ }
142+ setAuthHeader(req, auth)
143+
144+ resp, err := p.httpClient().Do(req)
145+ if err != nil {
146+ return "", fmt.Errorf("failed to start blob upload: %w", err)
147+ }
148+ defer resp.Body.Close()
149+
150+ if resp.StatusCode != http.StatusAccepted {
151+ return "", fmt.Errorf("blob upload start returned status %d", resp.StatusCode)
152+ }
153+
154+ location, err := resolveLocation(baseURL, resp.Header.Get("Location"))
155+ if err != nil {
156+ return "", fmt.Errorf("failed to resolve blob upload location: %w", err)
157+ }
158+
159+ return location, nil
160+}
161+
162+func (p *Pusher) uploadBlobChunks(ctx context.Context, uploadURL string, content io.Reader, auth string) (string, error) {
163+ buffer := make([]byte, p.chunkSize)
164+ offset := int64(0)
165+
166+ for {
167+ readBytes, err := io.ReadFull(io.LimitReader(content, p.chunkSize), buffer)
168+ if err == io.EOF {
169+ return uploadURL, nil
170+ }
171+ if err != nil && err != io.ErrUnexpectedEOF {
172+ return "", fmt.Errorf("failed to read blob chunk: %w", err)
173+ }
174+
175+ nextURL, patchErr := p.uploadBlobChunk(ctx, uploadURL, buffer[:readBytes], offset, auth)
176+ if patchErr != nil {
177+ return "", patchErr
178+ }
179+
180+ uploadURL = nextURL
181+ offset += int64(readBytes)
182+
183+ if err == io.ErrUnexpectedEOF {
184+ return uploadURL, nil
185+ }
186+ }
187+}
188+
189+func (p *Pusher) uploadBlobChunk(ctx context.Context, uploadURL string, chunk []byte, offset int64, auth string) (string, error) {
190+ req, err := http.NewRequestWithContext(ctx, http.MethodPatch, uploadURL, io.NopCloser(io.LimitReader(bytesReader(chunk), int64(len(chunk)))))
191+ if err != nil {
192+ return "", fmt.Errorf("failed to create blob chunk request: %w", err)
193+ }
194+ setAuthHeader(req, auth)
195+ req.Header.Set("Content-Type", "application/octet-stream")
196+ req.Header.Set("Content-Range", fmt.Sprintf("%d-%d", offset, offset+int64(len(chunk))-1))
197+ req.ContentLength = int64(len(chunk))
198+
199+ resp, err := p.httpClient().Do(req)
200+ if err != nil {
201+ return "", fmt.Errorf("failed to upload blob chunk: %w", err)
202+ }
203+ defer resp.Body.Close()
204+
205+ if resp.StatusCode != http.StatusAccepted {
206+ return "", fmt.Errorf("blob chunk upload returned status %d", resp.StatusCode)
207+ }
208+
209+ nextURL, err := resolveLocation(uploadURL, resp.Header.Get("Location"))
210+ if err != nil {
211+ return "", fmt.Errorf("failed to resolve next blob upload location: %w", err)
212+ }
213+
214+ return nextURL, nil
215+}
216+
217+func (p *Pusher) finalizeBlobUpload(ctx context.Context, uploadURL, digest, auth string) error {
218+ parsedURL, err := url.Parse(uploadURL)
219+ if err != nil {
220+ return fmt.Errorf("failed to parse blob upload location: %w", err)
221+ }
222+
223+ query := parsedURL.Query()
224+ query.Set("digest", digest)
225+ parsedURL.RawQuery = query.Encode()
226+
227+ req, err := http.NewRequestWithContext(ctx, http.MethodPut, parsedURL.String(), nil)
228+ if err != nil {
229+ return fmt.Errorf("failed to create blob finalize request: %w", err)
230+ }
231+ setAuthHeader(req, auth)
232+
233+ resp, err := p.httpClient().Do(req)
234+ if err != nil {
235+ return fmt.Errorf("failed to finalize blob upload: %w", err)
236+ }
237+ defer resp.Body.Close()
238+
239+ if resp.StatusCode != http.StatusCreated {
240+ return fmt.Errorf("blob upload finalize returned status %d", resp.StatusCode)
241+ }
242+
243+ return nil
244+}
245+
246+func (p *Pusher) httpClient() *http.Client {
247+ return &http.Client{Transport: p.transport}
248+}
249+
250+func setAuthHeader(req *http.Request, auth string) {
251+ if auth != "" {
252+ req.Header.Set("Authorization", auth)
253+ }
254+}
255+
256+func resolveLocation(baseURL, location string) (string, error) {
257+ if location == "" {
258+ return "", fmt.Errorf("missing Location header")
259+ }
260+
261+ base, err := url.Parse(baseURL)
262+ if err != nil {
263+ return "", err
264+ }
265+
266+ ref, err := url.Parse(location)
267+ if err != nil {
268+ return "", err
269+ }
270+
271+ return base.ResolveReference(ref).String(), nil
272+}
273+
274+func chunkCount(size, chunkSize int64) int64 {
275+ if size == 0 || chunkSize <= 0 {
276+ return 0
277+ }
278+ return (size + chunkSize - 1) / chunkSize
279+}
280+
281+func bytesReader(chunk []byte) io.Reader {
282+ return &sliceReader{chunk: chunk}
283+}
284+
285+type sliceReader struct {
286+ chunk []byte
287+ off int
288+}
289+
290+func (r *sliceReader) Read(p []byte) (int, error) {
291+ if r.off >= len(r.chunk) {
292+ return 0, io.EOF
293+ }
294+ n := copy(p, r.chunk[r.off:])
295+ r.off += n
296+ return n, nil
297+}
@@ -2,8 +2,10 @@ package registrypush
2 2
3 import (3 import (
4 "context"4 "context"
5+ "fmt"
5 "io"6 "io"
6 "net/http"7 "net/http"
8+ "net/url"
7 9
8 v1 "github.com/google/go-containerregistry/pkg/v1"10 v1 "github.com/google/go-containerregistry/pkg/v1"
9 )11 )
@@ -72,8 +74,224 @@ func (p *Pusher) Push(ctx context.Context, ref string) error {
72 return nil74 return nil
73 }75 }
74 76
77+// UploadBlob uploads a single blob to the registry using chunked uploads.
78+// If the blob already exists, it is skipped.
79+func (p *Pusher) UploadBlob(ctx context.Context, baseURL, repo, digest string, size int64, content io.Reader, auth string) error {
80+ if err := p.checkBlobExists(ctx, baseURL, repo, digest, auth); err == nil {
81+ return nil
82+ } else if err != errBlobMissing {
83+ return err
84+ }
85+
86+ uploadURL, err := p.startBlobUpload(ctx, baseURL, repo, auth)
87+ if err != nil {
88+ return err
89+ }
90+
91+ chunks := chunkCount(size, p.chunkSize)
92+ if p.progress != nil {
93+ if _, err := fmt.Fprintf(p.progress, "Uploading blob %s (%d bytes, %d chunks)\n", digest, size, chunks); err != nil {
94+ return fmt.Errorf("failed to write upload progress: %w", err)
95+ }
96+ }
97+
98+ uploadURL, err = p.uploadBlobChunks(ctx, uploadURL, content, auth)
99+ if err != nil {
100+ return err
101+ }
102+
103+ return p.finalizeBlobUpload(ctx, uploadURL, digest, auth)
104+}
105+
75 // defaultImageSource reads an image from the local Docker/Podman daemon.106 // defaultImageSource reads an image from the local Docker/Podman daemon.
76 func defaultImageSource(ctx context.Context, ref string) (v1.Image, error) {107 func defaultImageSource(ctx context.Context, ref string) (v1.Image, error) {
77 // Implementation in Task 3108 // Implementation in Task 3
78 return nil, nil109 return nil, nil
79 }110 }
111+
112+var errBlobMissing = fmt.Errorf("blob missing")
113+
114+func (p *Pusher) checkBlobExists(ctx context.Context, baseURL, repo, digest, auth string) error {
115+ req, err := http.NewRequestWithContext(ctx, http.MethodHead, fmt.Sprintf("%s/v2/%s/blobs/%s", baseURL, repo, digest), nil)
116+ if err != nil {
117+ return fmt.Errorf("failed to create blob HEAD request: %w", err)
118+ }
119+ setAuthHeader(req, auth)
120+
121+ resp, err := p.httpClient().Do(req)
122+ if err != nil {
123+ return fmt.Errorf("failed to check blob existence: %w", err)
124+ }
125+ defer resp.Body.Close()
126+
127+ switch resp.StatusCode {
128+ case http.StatusOK:
129+ return nil
130+ case http.StatusNotFound:
131+ return errBlobMissing
132+ default:
133+ return fmt.Errorf("blob existence check returned status %d", resp.StatusCode)
134+ }
135+}
136+
137+func (p *Pusher) startBlobUpload(ctx context.Context, baseURL, repo, auth string) (string, error) {
138+ req, err := http.NewRequestWithContext(ctx, http.MethodPost, fmt.Sprintf("%s/v2/%s/blobs/uploads/", baseURL, repo), nil)
139+ if err != nil {
140+ return "", fmt.Errorf("failed to create blob upload start request: %w", err)
141+ }
142+ setAuthHeader(req, auth)
143+
144+ resp, err := p.httpClient().Do(req)
145+ if err != nil {
146+ return "", fmt.Errorf("failed to start blob upload: %w", err)
147+ }
148+ defer resp.Body.Close()
149+
150+ if resp.StatusCode != http.StatusAccepted {
151+ return "", fmt.Errorf("blob upload start returned status %d", resp.StatusCode)
152+ }
153+
154+ location, err := resolveLocation(baseURL, resp.Header.Get("Location"))
155+ if err != nil {
156+ return "", fmt.Errorf("failed to resolve blob upload location: %w", err)
157+ }
158+
159+ return location, nil
160+}
161+
162+func (p *Pusher) uploadBlobChunks(ctx context.Context, uploadURL string, content io.Reader, auth string) (string, error) {
163+ buffer := make([]byte, p.chunkSize)
164+ offset := int64(0)
165+
166+ for {
167+ readBytes, err := io.ReadFull(io.LimitReader(content, p.chunkSize), buffer)
168+ if err == io.EOF {
169+ return uploadURL, nil
170+ }
171+ if err != nil && err != io.ErrUnexpectedEOF {
172+ return "", fmt.Errorf("failed to read blob chunk: %w", err)
173+ }
174+
175+ nextURL, patchErr := p.uploadBlobChunk(ctx, uploadURL, buffer[:readBytes], offset, auth)
176+ if patchErr != nil {
177+ return "", patchErr
178+ }
179+
180+ uploadURL = nextURL
181+ offset += int64(readBytes)
182+
183+ if err == io.ErrUnexpectedEOF {
184+ return uploadURL, nil
185+ }
186+ }
187+}
188+
189+func (p *Pusher) uploadBlobChunk(ctx context.Context, uploadURL string, chunk []byte, offset int64, auth string) (string, error) {
190+ req, err := http.NewRequestWithContext(ctx, http.MethodPatch, uploadURL, io.NopCloser(io.LimitReader(bytesReader(chunk), int64(len(chunk)))))
191+ if err != nil {
192+ return "", fmt.Errorf("failed to create blob chunk request: %w", err)
193+ }
194+ setAuthHeader(req, auth)
195+ req.Header.Set("Content-Type", "application/octet-stream")
196+ req.Header.Set("Content-Range", fmt.Sprintf("%d-%d", offset, offset+int64(len(chunk))-1))
197+ req.ContentLength = int64(len(chunk))
198+
199+ resp, err := p.httpClient().Do(req)
200+ if err != nil {
201+ return "", fmt.Errorf("failed to upload blob chunk: %w", err)
202+ }
203+ defer resp.Body.Close()
204+
205+ if resp.StatusCode != http.StatusAccepted {
206+ return "", fmt.Errorf("blob chunk upload returned status %d", resp.StatusCode)
207+ }
208+
209+ nextURL, err := resolveLocation(uploadURL, resp.Header.Get("Location"))
210+ if err != nil {
211+ return "", fmt.Errorf("failed to resolve next blob upload location: %w", err)
212+ }
213+
214+ return nextURL, nil
215+}
216+
217+func (p *Pusher) finalizeBlobUpload(ctx context.Context, uploadURL, digest, auth string) error {
218+ parsedURL, err := url.Parse(uploadURL)
219+ if err != nil {
220+ return fmt.Errorf("failed to parse blob upload location: %w", err)
221+ }
222+
223+ query := parsedURL.Query()
224+ query.Set("digest", digest)
225+ parsedURL.RawQuery = query.Encode()
226+
227+ req, err := http.NewRequestWithContext(ctx, http.MethodPut, parsedURL.String(), nil)
228+ if err != nil {
229+ return fmt.Errorf("failed to create blob finalize request: %w", err)
230+ }
231+ setAuthHeader(req, auth)
232+
233+ resp, err := p.httpClient().Do(req)
234+ if err != nil {
235+ return fmt.Errorf("failed to finalize blob upload: %w", err)
236+ }
237+ defer resp.Body.Close()
238+
239+ if resp.StatusCode != http.StatusCreated {
240+ return fmt.Errorf("blob upload finalize returned status %d", resp.StatusCode)
241+ }
242+
243+ return nil
244+}
245+
246+func (p *Pusher) httpClient() *http.Client {
247+ return &http.Client{Transport: p.transport}
248+}
249+
250+func setAuthHeader(req *http.Request, auth string) {
251+ if auth != "" {
252+ req.Header.Set("Authorization", auth)
253+ }
254+}
255+
256+func resolveLocation(baseURL, location string) (string, error) {
257+ if location == "" {
258+ return "", fmt.Errorf("missing Location header")
259+ }
260+
261+ base, err := url.Parse(baseURL)
262+ if err != nil {
263+ return "", err
264+ }
265+
266+ ref, err := url.Parse(location)
267+ if err != nil {
268+ return "", err
269+ }
270+
271+ return base.ResolveReference(ref).String(), nil
272+}
273+
274+func chunkCount(size, chunkSize int64) int64 {
275+ if size == 0 || chunkSize <= 0 {
276+ return 0
277+ }
278+ return (size + chunkSize - 1) / chunkSize
279+}
280+
281+func bytesReader(chunk []byte) io.Reader {
282+ return &sliceReader{chunk: chunk}
283+}
284+
285+type sliceReader struct {
286+ chunk []byte
287+ off int
288+}
289+
290+func (r *sliceReader) Read(p []byte) (int, error) {
291+ if r.off >= len(r.chunk) {
292+ return 0, io.EOF
293+ }
294+ n := copy(p, r.chunk[r.off:])
295+ r.off += n
296+ return n, nil
297+}
M pkg/registrypush/push_test.go +209 -0
@@ -1,9 +1,18 @@
11 package registrypush_test
22
33 import (
4+ "bytes"
5+ "context"
6+ "fmt"
7+ "io"
8+ "net/http"
9+ "net/http/httptest"
10+ "strconv"
11+ "strings"
412 "testing"
513
614 "github.com/stretchr/testify/assert"
15+ "github.com/stretchr/testify/require"
716
817 "github.com/bnema/gordon/pkg/registrypush"
918 )
@@ -17,3 +26,203 @@ func TestNew_withChunkSize(t *testing.T) {
1726 p := registrypush.New(registrypush.WithChunkSize(10 * 1024 * 1024))
1827 assert.NotNil(t, p)
1928 }
29+
30+func TestPusher_UploadBlob(t *testing.T) {
31+ testCases := []struct {
32+ name string
33+ blobSize int
34+ chunkSize int64
35+ headStatus int
36+ expectedPatchSizes []int
37+ expectedHeadCalls int
38+ expectedPostCalls int
39+ expectedPutCalls int
40+ }{
41+ {
42+ name: "small blob under chunk size",
43+ blobSize: 100,
44+ chunkSize: 1024,
45+ headStatus: http.StatusNotFound,
46+ expectedPatchSizes: []int{100},
47+ expectedHeadCalls: 1,
48+ expectedPostCalls: 1,
49+ expectedPutCalls: 1,
50+ },
51+ {
52+ name: "exact chunk size",
53+ blobSize: 1024,
54+ chunkSize: 1024,
55+ headStatus: http.StatusNotFound,
56+ expectedPatchSizes: []int{1024},
57+ expectedHeadCalls: 1,
58+ expectedPostCalls: 1,
59+ expectedPutCalls: 1,
60+ },
61+ {
62+ name: "large blob across multiple chunks",
63+ blobSize: 3072,
64+ chunkSize: 1024,
65+ headStatus: http.StatusNotFound,
66+ expectedPatchSizes: []int{1024, 1024, 1024},
67+ expectedHeadCalls: 1,
68+ expectedPostCalls: 1,
69+ expectedPutCalls: 1,
70+ },
71+ {
72+ name: "blob already exists",
73+ blobSize: 512,
74+ chunkSize: 1024,
75+ headStatus: http.StatusOK,
76+ expectedPatchSizes: nil,
77+ expectedHeadCalls: 1,
78+ expectedPostCalls: 0,
79+ expectedPutCalls: 0,
80+ },
81+ {
82+ name: "auth header present on every request",
83+ blobSize: 1536,
84+ chunkSize: 1024,
85+ headStatus: http.StatusNotFound,
86+ expectedPatchSizes: []int{1024, 512},
87+ expectedHeadCalls: 1,
88+ expectedPostCalls: 1,
89+ expectedPutCalls: 1,
90+ },
91+ }
92+
93+ for _, tc := range testCases {
94+ t.Run(tc.name, func(t *testing.T) {
95+ const authHeader = "Bearer test-token"
96+ const repo = "demo/app"
97+ const digest = "sha256:testdigest"
98+
99+ blob := bytes.Repeat([]byte("a"), tc.blobSize)
100+ serverState := newFakeBlobUploadRegistry(t, authHeader, digest, tc.headStatus)
101+ server := httptest.NewServer(serverState.handler())
102+ defer server.Close()
103+
104+ p := registrypush.New(
105+ registrypush.WithChunkSize(tc.chunkSize),
106+ registrypush.WithTransport(server.Client().Transport),
107+ )
108+
109+ err := p.UploadBlob(context.Background(), server.URL, repo, digest, int64(len(blob)), bytes.NewReader(blob), authHeader)
110+ require.NoError(t, err)
111+
112+ assert.Equal(t, tc.expectedHeadCalls, serverState.headCalls)
113+ assert.Equal(t, tc.expectedPostCalls, serverState.postCalls)
114+ assert.Equal(t, tc.expectedPutCalls, serverState.putCalls)
115+ assert.Equal(t, tc.expectedPatchSizes, serverState.patchSizes)
116+ assert.Equal(t, tc.expectedPatchSizes, serverState.rangeSizes)
117+ assert.Equal(t, serverState.requestCount, serverState.authCount)
118+ })
119+ }
120+}
121+
122+type fakeBlobUploadRegistry struct {
123+ t *testing.T
124+ expectedAuth string
125+ expectedDigest string
126+ headStatus int
127+ patchSizes []int
128+ rangeSizes []int
129+ headCalls int
130+ postCalls int
131+ putCalls int
132+ requestCount int
133+ authCount int
134+ nextOffset int
135+ location string
136+ uuid string
137+}
138+
139+func newFakeBlobUploadRegistry(t *testing.T, expectedAuth, expectedDigest string, headStatus int) *fakeBlobUploadRegistry {
140+ t.Helper()
141+ return &fakeBlobUploadRegistry{
142+ t: t,
143+ expectedAuth: expectedAuth,
144+ expectedDigest: expectedDigest,
145+ headStatus: headStatus,
146+ location: "/upload/session-1",
147+ uuid: "session-1",
148+ }
149+}
150+
151+func (f *fakeBlobUploadRegistry) handler() http.Handler {
152+ f.t.Helper()
153+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
154+ f.requestCount++
155+ if r.Header.Get("Authorization") == f.expectedAuth {
156+ f.authCount++
157+ }
158+
159+ switch {
160+ case r.Method == http.MethodHead && strings.Contains(r.URL.Path, "/blobs/"):
161+ f.headCalls++
162+ w.WriteHeader(f.headStatus)
163+ case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/blobs/uploads/"):
164+ f.postCalls++
165+ w.Header().Set("Location", f.location)
166+ w.Header().Set("Docker-Upload-UUID", f.uuid)
167+ w.Header().Set("Range", rangeHeaderValue(0, f.nextOffset-1))
168+ w.WriteHeader(http.StatusAccepted)
169+ case r.Method == http.MethodPatch && r.URL.Path == f.location:
170+ f.handlePatch(w, r)
171+ case r.Method == http.MethodPut && r.URL.Path == f.location:
172+ f.putCalls++
173+ assert.Equal(f.t, f.expectedDigest, r.URL.Query().Get("digest"))
174+ body, err := io.ReadAll(r.Body)
175+ require.NoError(f.t, err)
176+ assert.Len(f.t, body, 0)
177+ w.Header().Set("Location", f.location)
178+ w.Header().Set("Docker-Content-Digest", f.expectedDigest)
179+ w.WriteHeader(http.StatusCreated)
180+ default:
181+ f.t.Fatalf("unexpected request: %s %s", r.Method, r.URL.String())
182+ }
183+ })
184+}
185+
186+func (f *fakeBlobUploadRegistry) handlePatch(w http.ResponseWriter, r *http.Request) {
187+ body, err := io.ReadAll(r.Body)
188+ require.NoError(f.t, err)
189+
190+ start, end := parseContentRange(f.t, r.Header.Get("Content-Range"))
191+ assert.Equal(f.t, f.nextOffset, start)
192+ assert.Equal(f.t, len(body), end-start+1)
193+
194+ contentLength, err := strconv.Atoi(r.Header.Get("Content-Length"))
195+ require.NoError(f.t, err)
196+ assert.Equal(f.t, len(body), contentLength)
197+ assert.Equal(f.t, "application/octet-stream", r.Header.Get("Content-Type"))
198+
199+ f.patchSizes = append(f.patchSizes, len(body))
200+ f.rangeSizes = append(f.rangeSizes, end-start+1)
201+ f.nextOffset = end + 1
202+
203+ w.Header().Set("Location", f.location)
204+ w.Header().Set("Docker-Upload-UUID", f.uuid)
205+ w.Header().Set("Range", rangeHeaderValue(0, end))
206+ w.WriteHeader(http.StatusAccepted)
207+}
208+
209+func parseContentRange(t *testing.T, value string) (int, int) {
210+ t.Helper()
211+ parts := strings.Split(value, "-")
212+ require.Len(t, parts, 2)
213+
214+ start, err := strconv.Atoi(parts[0])
215+ require.NoError(t, err)
216+
217+ end, err := strconv.Atoi(parts[1])
218+ require.NoError(t, err)
219+
220+ return start, end
221+}
222+
223+func rangeHeaderValue(start, end int) string {
224+ if end < start {
225+ return ""
226+ }
227+ return fmt.Sprintf("%d-%d", start, end)
228+}
@@ -1,9 +1,18 @@
1 package registrypush_test1 package registrypush_test
2 2
3 import (3 import (
4+ "bytes"
5+ "context"
6+ "fmt"
7+ "io"
8+ "net/http"
9+ "net/http/httptest"
10+ "strconv"
11+ "strings"
4 "testing"12 "testing"
5 13
6 "github.com/stretchr/testify/assert"14 "github.com/stretchr/testify/assert"
15+ "github.com/stretchr/testify/require"
7 16
8 "github.com/bnema/gordon/pkg/registrypush"17 "github.com/bnema/gordon/pkg/registrypush"
9 )18 )
@@ -17,3 +26,203 @@ func TestNew_withChunkSize(t *testing.T) {
17 p := registrypush.New(registrypush.WithChunkSize(10 * 1024 * 1024))26 p := registrypush.New(registrypush.WithChunkSize(10 * 1024 * 1024))
18 assert.NotNil(t, p)27 assert.NotNil(t, p)
19 }28 }
29+
30+func TestPusher_UploadBlob(t *testing.T) {
31+ testCases := []struct {
32+ name string
33+ blobSize int
34+ chunkSize int64
35+ headStatus int
36+ expectedPatchSizes []int
37+ expectedHeadCalls int
38+ expectedPostCalls int
39+ expectedPutCalls int
40+ }{
41+ {
42+ name: "small blob under chunk size",
43+ blobSize: 100,
44+ chunkSize: 1024,
45+ headStatus: http.StatusNotFound,
46+ expectedPatchSizes: []int{100},
47+ expectedHeadCalls: 1,
48+ expectedPostCalls: 1,
49+ expectedPutCalls: 1,
50+ },
51+ {
52+ name: "exact chunk size",
53+ blobSize: 1024,
54+ chunkSize: 1024,
55+ headStatus: http.StatusNotFound,
56+ expectedPatchSizes: []int{1024},
57+ expectedHeadCalls: 1,
58+ expectedPostCalls: 1,
59+ expectedPutCalls: 1,
60+ },
61+ {
62+ name: "large blob across multiple chunks",
63+ blobSize: 3072,
64+ chunkSize: 1024,
65+ headStatus: http.StatusNotFound,
66+ expectedPatchSizes: []int{1024, 1024, 1024},
67+ expectedHeadCalls: 1,
68+ expectedPostCalls: 1,
69+ expectedPutCalls: 1,
70+ },
71+ {
72+ name: "blob already exists",
73+ blobSize: 512,
74+ chunkSize: 1024,
75+ headStatus: http.StatusOK,
76+ expectedPatchSizes: nil,
77+ expectedHeadCalls: 1,
78+ expectedPostCalls: 0,
79+ expectedPutCalls: 0,
80+ },
81+ {
82+ name: "auth header present on every request",
83+ blobSize: 1536,
84+ chunkSize: 1024,
85+ headStatus: http.StatusNotFound,
86+ expectedPatchSizes: []int{1024, 512},
87+ expectedHeadCalls: 1,
88+ expectedPostCalls: 1,
89+ expectedPutCalls: 1,
90+ },
91+ }
92+
93+ for _, tc := range testCases {
94+ t.Run(tc.name, func(t *testing.T) {
95+ const authHeader = "Bearer test-token"
96+ const repo = "demo/app"
97+ const digest = "sha256:testdigest"
98+
99+ blob := bytes.Repeat([]byte("a"), tc.blobSize)
100+ serverState := newFakeBlobUploadRegistry(t, authHeader, digest, tc.headStatus)
101+ server := httptest.NewServer(serverState.handler())
102+ defer server.Close()
103+
104+ p := registrypush.New(
105+ registrypush.WithChunkSize(tc.chunkSize),
106+ registrypush.WithTransport(server.Client().Transport),
107+ )
108+
109+ err := p.UploadBlob(context.Background(), server.URL, repo, digest, int64(len(blob)), bytes.NewReader(blob), authHeader)
110+ require.NoError(t, err)
111+
112+ assert.Equal(t, tc.expectedHeadCalls, serverState.headCalls)
113+ assert.Equal(t, tc.expectedPostCalls, serverState.postCalls)
114+ assert.Equal(t, tc.expectedPutCalls, serverState.putCalls)
115+ assert.Equal(t, tc.expectedPatchSizes, serverState.patchSizes)
116+ assert.Equal(t, tc.expectedPatchSizes, serverState.rangeSizes)
117+ assert.Equal(t, serverState.requestCount, serverState.authCount)
118+ })
119+ }
120+}
121+
122+type fakeBlobUploadRegistry struct {
123+ t *testing.T
124+ expectedAuth string
125+ expectedDigest string
126+ headStatus int
127+ patchSizes []int
128+ rangeSizes []int
129+ headCalls int
130+ postCalls int
131+ putCalls int
132+ requestCount int
133+ authCount int
134+ nextOffset int
135+ location string
136+ uuid string
137+}
138+
139+func newFakeBlobUploadRegistry(t *testing.T, expectedAuth, expectedDigest string, headStatus int) *fakeBlobUploadRegistry {
140+ t.Helper()
141+ return &fakeBlobUploadRegistry{
142+ t: t,
143+ expectedAuth: expectedAuth,
144+ expectedDigest: expectedDigest,
145+ headStatus: headStatus,
146+ location: "/upload/session-1",
147+ uuid: "session-1",
148+ }
149+}
150+
151+func (f *fakeBlobUploadRegistry) handler() http.Handler {
152+ f.t.Helper()
153+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
154+ f.requestCount++
155+ if r.Header.Get("Authorization") == f.expectedAuth {
156+ f.authCount++
157+ }
158+
159+ switch {
160+ case r.Method == http.MethodHead && strings.Contains(r.URL.Path, "/blobs/"):
161+ f.headCalls++
162+ w.WriteHeader(f.headStatus)
163+ case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/blobs/uploads/"):
164+ f.postCalls++
165+ w.Header().Set("Location", f.location)
166+ w.Header().Set("Docker-Upload-UUID", f.uuid)
167+ w.Header().Set("Range", rangeHeaderValue(0, f.nextOffset-1))
168+ w.WriteHeader(http.StatusAccepted)
169+ case r.Method == http.MethodPatch && r.URL.Path == f.location:
170+ f.handlePatch(w, r)
171+ case r.Method == http.MethodPut && r.URL.Path == f.location:
172+ f.putCalls++
173+ assert.Equal(f.t, f.expectedDigest, r.URL.Query().Get("digest"))
174+ body, err := io.ReadAll(r.Body)
175+ require.NoError(f.t, err)
176+ assert.Len(f.t, body, 0)
177+ w.Header().Set("Location", f.location)
178+ w.Header().Set("Docker-Content-Digest", f.expectedDigest)
179+ w.WriteHeader(http.StatusCreated)
180+ default:
181+ f.t.Fatalf("unexpected request: %s %s", r.Method, r.URL.String())
182+ }
183+ })
184+}
185+
186+func (f *fakeBlobUploadRegistry) handlePatch(w http.ResponseWriter, r *http.Request) {
187+ body, err := io.ReadAll(r.Body)
188+ require.NoError(f.t, err)
189+
190+ start, end := parseContentRange(f.t, r.Header.Get("Content-Range"))
191+ assert.Equal(f.t, f.nextOffset, start)
192+ assert.Equal(f.t, len(body), end-start+1)
193+
194+ contentLength, err := strconv.Atoi(r.Header.Get("Content-Length"))
195+ require.NoError(f.t, err)
196+ assert.Equal(f.t, len(body), contentLength)
197+ assert.Equal(f.t, "application/octet-stream", r.Header.Get("Content-Type"))
198+
199+ f.patchSizes = append(f.patchSizes, len(body))
200+ f.rangeSizes = append(f.rangeSizes, end-start+1)
201+ f.nextOffset = end + 1
202+
203+ w.Header().Set("Location", f.location)
204+ w.Header().Set("Docker-Upload-UUID", f.uuid)
205+ w.Header().Set("Range", rangeHeaderValue(0, end))
206+ w.WriteHeader(http.StatusAccepted)
207+}
208+
209+func parseContentRange(t *testing.T, value string) (int, int) {
210+ t.Helper()
211+ parts := strings.Split(value, "-")
212+ require.Len(t, parts, 2)
213+
214+ start, err := strconv.Atoi(parts[0])
215+ require.NoError(t, err)
216+
217+ end, err := strconv.Atoi(parts[1])
218+ require.NoError(t, err)
219+
220+ return start, end
221+}
222+
223+func rangeHeaderValue(start, end int) string {
224+ if end < start {
225+ return ""
226+ }
227+ return fmt.Sprintf("%d-%d", start, end)
228+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 8

Tags 106

↑↓ moveenter openesc close