| index 93edf1c67..edb13047f 100644 |
| --- a/internal/release/store.go |
| +++ b/internal/release/store.go |
| @@ -134,6 +134,12 @@ func (p prefixStore) Put(ctx context.Context, key string, body []byte, ifMatch s |
| return p.Store.Put(ctx, p.full(key), body, ifMatch) |
| } |
| |
| +// PutFrom is written out, not promoted, for the reason Open is: a promoted |
| +// method would forward the UNREBASED key. |
| +func (p prefixStore) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) { |
| + return PutObjectFrom(ctx, p.Store, p.full(key), body, size, ifMatch) |
| +} |
| + |
| func (p prefixStore) Delete(ctx context.Context, key string) error { |
| return p.Store.Delete(ctx, p.full(key)) |
| } |
| @@ -229,10 +235,51 @@ type Opener interface { |
| |
| var ( |
| _ Opener = (*DirStore)(nil) |
| + |
| _ Opener = (*S3Store)(nil) |
| _ Opener = prefixStore{} |
| ) |
| |
| +// PutStreamer is the optional streaming write: Store.Put takes the whole |
| +// body as []byte, which is wrong for a 130 MB GeoIP database on a box |
| +// under MemoryHigh. Implemented by every production store; a fake with only |
| +// Put still works through PutObjectFrom for bodies up to PutFromFallbackMax. |
| +// ifMatch has Put's exact semantics ("" = create-only). |
| +type PutStreamer interface { |
| + PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) |
| +} |
| + |
| +var ( |
| + _ PutStreamer = (*DirStore)(nil) |
| + _ PutStreamer = (*S3Store)(nil) |
| + _ PutStreamer = prefixStore{} |
| +) |
| + |
| +// PutFromFallbackMax bounds how much PutObjectFrom will buffer for a store |
| +// with no PutFrom. |
| +const PutFromFallbackMax = 64 << 20 |
| + |
| +// ErrTooLargeForPut is PutObjectFrom refusing to buffer a large body. |
| +var ErrTooLargeForPut = errors.New("store cannot stream and the body is too large to buffer") |
| + |
| +// PutObjectFrom streams body into s at key. |
| +func PutObjectFrom(ctx context.Context, s Store, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) { |
| + if ps, ok := s.(PutStreamer); ok { |
| + return ps.PutFrom(ctx, key, body, size, ifMatch) |
| + } |
| + if size > PutFromFallbackMax { |
| + return "", fmt.Errorf("%s: %w", key, ErrTooLargeForPut) |
| + } |
| + b, err := io.ReadAll(io.LimitReader(body, size+1)) |
| + if err != nil { |
| + return "", err |
| + } |
| + if int64(len(b)) != size { |
| + return "", fmt.Errorf("%s: body is %d bytes, want %d", key, len(b), size) |
| + } |
| + return s.Put(ctx, key, b, ifMatch) |
| +} |
| + |
| // OpenObject streams key from s, buffering through Get when s cannot stream. |
| func OpenObject(ctx context.Context, s Store, key string) (io.ReadCloser, string, error) { |
| if o, ok := s.(Opener); ok { |
| @@ -373,6 +420,55 @@ func (d *DirStore) Put(_ context.Context, key string, body []byte, ifMatch strin |
| return etagOf(body), nil |
| } |
| |
| +// PutFrom is Put streamed: the body goes to a temp file (hashed as it |
| +// goes, so the etag is the same content sha256 Put would give) and is |
| +// renamed into place under d.mu after the same CAS gate. |
| +func (d *DirStore) PutFrom(_ context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) { |
| + p, err := d.path(key) |
| + if err != nil { |
| + return "", err |
| + } |
| + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { |
| + return "", err |
| + } |
| + tmp, err := os.CreateTemp(filepath.Dir(p), filepath.Base(p)+".*.tmp") |
| + if err != nil { |
| + return "", err |
| + } |
| + defer os.Remove(tmp.Name()) |
| + h := sha256.New() |
| + n, err := io.Copy(io.MultiWriter(tmp, h), io.LimitReader(body, size+1)) |
| + if cerr := tmp.Close(); err == nil { |
| + err = cerr |
| + } |
| + if err != nil { |
| + return "", err |
| + } |
| + if n != size { |
| + return "", fmt.Errorf("dirstore: %s: body is %d bytes, want %d", key, n, size) |
| + } |
| + d.mu.Lock() |
| + defer d.mu.Unlock() |
| + cur, err := os.ReadFile(p) |
| + exists := err == nil |
| + if err != nil && !errors.Is(err, fs.ErrNotExist) { |
| + return "", err |
| + } |
| + if ifMatch == "" && exists { |
| + return "", fmt.Errorf("dirstore: %s exists: %w", key, ErrCASConflict) |
| + } |
| + if ifMatch != "" && (!exists || etagOf(cur) != ifMatch) { |
| + return "", fmt.Errorf("dirstore: %s moved on: %w", key, ErrCASConflict) |
| + } |
| + if err := os.Chmod(tmp.Name(), 0o644); err != nil { |
| + return "", err |
| + } |
| + if err := os.Rename(tmp.Name(), p); err != nil { |
| + return "", err |
| + } |
| + return hex.EncodeToString(h.Sum(nil)), nil |
| +} |
| + |
| // List is ListMod minus the mtimes — one walk owns the prefix filter and the |
| // .tmp exclusion, so the two listings cannot drift. |
| func (d *DirStore) List(ctx context.Context, prefix string) ([]string, error) { |
| @@ -666,6 +762,29 @@ func (s *S3Store) Put(ctx context.Context, key string, body []byte, ifMatch stri |
| return aws.ToString(out.ETag), nil |
| } |
| |
| +// PutFrom is Put with a streamed, seekable body and an explicit length. |
| +func (s *S3Store) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) { |
| + in := &awss3.PutObjectInput{ |
| + Bucket: &s.Bucket, |
| + Key: &key, |
| + Body: body, |
| + ContentLength: aws.Int64(size), |
| + } |
| + if ifMatch == "" { |
| + in.IfNoneMatch = aws.String("*") |
| + } else { |
| + in.IfMatch = aws.String(ifMatch) |
| + } |
| + out, err := s.Client.PutObject(ctx, in) |
| + if err != nil { |
| + if s3CASConflict(err) || (ifMatch != "" && s3NotFound(err)) { |
| + return "", fmt.Errorf("s3store: %s: %w", key, ErrCASConflict) |
| + } |
| + return "", err |
| + } |
| + return aws.ToString(out.ETag), nil |
| +} |
| + |
| // List is ListMod minus the stamps — one paginator loop serves both. |
| func (s *S3Store) List(ctx context.Context, prefix string) ([]string, error) { |
| kms, err := s.ListMod(ctx, prefix) |
amadan