@@ -42,6 +42,9 @@ func New(ctx context.Context, spec remoteurl.Spec) (*Store, error) { |
| 42 |
client := s3.NewFromConfig(cfg, func(o *s3.Options) { |
42 |
client := s3.NewFromConfig(cfg, func(o *s3.Options) { |
| 43 |
o.BaseEndpoint = aws.String(spec.Endpoint) |
43 |
o.BaseEndpoint = aws.String(spec.Endpoint) |
| 44 |
o.UsePathStyle = true |
44 |
o.UsePathStyle = true |
|
|
45 |
// S3-compatible services do not uniformly support the SDK's optional |
|
|
46 |
// streaming CRC32 trailer, especially when Content-Encoding is set. |
|
|
47 |
o.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired |
| 45 |
}) |
48 |
}) |
| 46 |
return &Store{client: client, spec: spec}, nil |
49 |
return &Store{client: client, spec: spec}, nil |
| 47 |
} |
50 |
} |
@@ -74,35 +77,37 @@ func (s *Store) Get(ctx context.Context, name string) (Object, error) { |
| 74 |
} |
77 |
} |
| 75 |
|
78 |
|
| 76 |
func (s *Store) Put(ctx context.Context, name string, data []byte, contentType, cacheControl string) error { |
79 |
func (s *Store) Put(ctx context.Context, name string, data []byte, contentType, cacheControl string) error { |
| 77 |
input := &s3.PutObjectInput{ |
80 |
return s.putBytes(ctx, s.key(name), data, contentType, cacheControl, "") |
| 78 |
Bucket: aws.String(s.spec.Bucket), |
81 |
} |
| 79 |
Key: aws.String(s.key(name)), |
82 |
|
| 80 |
Body: bytes.NewReader(data), |
83 |
func (s *Store) PutEncoded(ctx context.Context, name string, data []byte, contentType, cacheControl, contentEncoding string) error { |
| 81 |
ContentType: aws.String(contentType), |
84 |
return s.putBytes(ctx, s.key(name), data, contentType, cacheControl, contentEncoding) |
| 82 |
} |
|
|
| 83 |
if cacheControl != "" { |
|
|
| 84 |
input.CacheControl = aws.String(cacheControl) |
|
|
| 85 |
} |
|
|
| 86 |
_, err := s.client.PutObject(ctx, input) |
|
|
| 87 |
if err != nil { |
|
|
| 88 |
return fmt.Errorf("put s3://%s/%s: %w", s.spec.Bucket, s.key(name), err) |
|
|
| 89 |
} |
|
|
| 90 |
return nil |
|
|
| 91 |
} |
85 |
} |
| 92 |
|
86 |
|
| 93 |
func (s *Store) PutRoot(ctx context.Context, data []byte, contentType, cacheControl string) error { |
87 |
func (s *Store) PutRoot(ctx context.Context, data []byte, contentType, cacheControl string) error { |
|
|
88 |
return s.putBytes(ctx, s.RootKey(), data, contentType, cacheControl, "") |
|
|
89 |
} |
|
|
90 |
|
|
|
91 |
func (s *Store) PutRootEncoded(ctx context.Context, data []byte, contentType, cacheControl, contentEncoding string) error { |
|
|
92 |
return s.putBytes(ctx, s.RootKey(), data, contentType, cacheControl, contentEncoding) |
|
|
93 |
} |
|
|
94 |
|
|
|
95 |
func (s *Store) putBytes(ctx context.Context, key string, data []byte, contentType, cacheControl, contentEncoding string) error { |
| 94 |
input := &s3.PutObjectInput{ |
96 |
input := &s3.PutObjectInput{ |
| 95 |
Bucket: aws.String(s.spec.Bucket), |
97 |
Bucket: aws.String(s.spec.Bucket), |
| 96 |
Key: aws.String(s.RootKey()), |
98 |
Key: aws.String(key), |
| 97 |
Body: bytes.NewReader(data), |
99 |
Body: bytes.NewReader(data), |
| 98 |
ContentType: aws.String(contentType), |
100 |
ContentType: aws.String(contentType), |
| 99 |
} |
101 |
} |
| 100 |
if cacheControl != "" { |
102 |
if cacheControl != "" { |
| 101 |
input.CacheControl = aws.String(cacheControl) |
103 |
input.CacheControl = aws.String(cacheControl) |
| 102 |
} |
104 |
} |
|
|
105 |
if contentEncoding != "" { |
|
|
106 |
input.ContentEncoding = aws.String(contentEncoding) |
|
|
107 |
} |
| 103 |
_, err := s.client.PutObject(ctx, input) |
108 |
_, err := s.client.PutObject(ctx, input) |
| 104 |
if err != nil { |
109 |
if err != nil { |
| 105 |
return fmt.Errorf("put s3://%s/%s: %w", s.spec.Bucket, s.RootKey(), err) |
110 |
return fmt.Errorf("put s3://%s/%s: %w", s.spec.Bucket, key, err) |
| 106 |
} |
111 |
} |
| 107 |
return nil |
112 |
return nil |
| 108 |
} |
113 |
} |
@@ -142,10 +147,11 @@ func (s *Store) Delete(ctx context.Context, name string) error { |
| 142 |
} |
147 |
} |
| 143 |
|
148 |
|
| 144 |
type Lock struct { |
149 |
type Lock struct { |
| 145 |
store *Store |
150 |
store *Store |
| 146 |
name string |
151 |
name string |
| 147 |
token string |
152 |
token string |
| 148 |
etag string |
153 |
etag string |
|
|
154 |
refreshedAt time.Time |
| 149 |
} |
155 |
} |
| 150 |
|
156 |
|
| 151 |
func (s *Store) AcquireLock(ctx context.Context) (*Lock, error) { |
157 |
func (s *Store) AcquireLock(ctx context.Context) (*Lock, error) { |
@@ -188,7 +194,7 @@ func (s *Store) AcquireLock(ctx context.Context) (*Lock, error) { |
| 188 |
return nil, fmt.Errorf("acquire repository lock: %w", err) |
194 |
return nil, fmt.Errorf("acquire repository lock: %w", err) |
| 189 |
} |
195 |
} |
| 190 |
} |
196 |
} |
| 191 |
return &Lock{store: s, name: name, token: token, etag: aws.ToString(out.ETag)}, nil |
197 |
return &Lock{store: s, name: name, token: token, etag: aws.ToString(out.ETag), refreshedAt: time.Now()}, nil |
| 192 |
} |
198 |
} |
| 193 |
|
199 |
|
| 194 |
// Refresh proves that this process still owns the lease before it publishes |
200 |
// Refresh proves that this process still owns the lease before it publishes |
@@ -206,9 +212,19 @@ func (l *Lock) Refresh(ctx context.Context) error { |
| 206 |
return fmt.Errorf("refresh repository lock: %w", err) |
212 |
return fmt.Errorf("refresh repository lock: %w", err) |
| 207 |
} |
213 |
} |
| 208 |
l.etag = aws.ToString(out.ETag) |
214 |
l.etag = aws.ToString(out.ETag) |
|
|
215 |
l.refreshedAt = time.Now() |
| 209 |
return nil |
216 |
return nil |
| 210 |
} |
217 |
} |
| 211 |
|
218 |
|
|
|
219 |
// RefreshIfOlderThan avoids an S3 round trip for ordinary pushes while still |
|
|
220 |
// renewing the lease during unusually long publications. |
|
|
221 |
func (l *Lock) RefreshIfOlderThan(ctx context.Context, age time.Duration) error { |
|
|
222 |
if time.Since(l.refreshedAt) < age { |
|
|
223 |
return nil |
|
|
224 |
} |
|
|
225 |
return l.Refresh(ctx) |
|
|
226 |
} |
|
|
227 |
|
| 212 |
func (l *Lock) Release(ctx context.Context) error { |
228 |
func (l *Lock) Release(ctx context.Context) error { |
| 213 |
_, err := l.store.client.DeleteObject(ctx, &s3.DeleteObjectInput{ |
229 |
_, err := l.store.client.DeleteObject(ctx, &s3.DeleteObjectInput{ |
| 214 |
Bucket: aws.String(l.store.spec.Bucket), |
230 |
Bucket: aws.String(l.store.spec.Bucket), |