1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3// Package blobdisk stores objects under root/objects, temporary writes under
4// root/tmp, and unfinished multipart uploads under root/multipart. Files are
5// 0600 in 0700 directories; completed objects appear by synced atomic rename.
6package blobdisk
7
8import (
9 "context"
10 "crypto/rand"
11 "crypto/sha256"
12 "encoding/hex"
13 "encoding/json"
14 "errors"
15 "fmt"
16 "io"
17 "io/fs"
18 "os"
19 "path/filepath"
20 "sort"
21 "strconv"
22 "strings"
23 "time"
24
25 "bnema.dev/ermite/internal/domain"
26 "bnema.dev/ermite/internal/ports"
27 "github.com/bnema/zerowrap"
28)
29
30// Store is the disk implementation of ports.BlobStore.
31type Store struct{ root string }
32
33var _ ports.BlobStore = (*Store)(nil)
34
35// Dir returns the blob root under dataDir.
36func Dir(dataDir string) string { return filepath.Join(dataDir, "blobs") }
37
38// New returns a disk store without performing I/O.
39func New(root string) *Store { return &Store{root: root} }
40func (s *Store) object(key string) string {
41 return filepath.Join(s.root, "objects", filepath.FromSlash(key))
42}
43func valid(key string) error {
44 if !ports.ValidBlobKey(key) {
45 return domain.ErrInvalid
46 }
47 return nil
48}
49func translate(err error) error {
50 if errors.Is(err, fs.ErrNotExist) {
51 return domain.ErrNotFound
52 }
53 return err
54}
55
56// tmpGrace is how long leftovers in root/tmp (and malformed multipart dirs)
57// are kept before reclamation.
58const tmpGrace = time.Hour
59
60// now is the clock used for reclamation; tests may replace it.
61var now = time.Now
62
63// syncDir fsyncs a directory; tests may replace it to record calls.
64var syncDir = func(path string) error {
65 f, e := os.Open(path)
66 if e != nil {
67 return e
68 }
69 return errors.Join(f.Sync(), f.Close())
70}
71
72// mkdirAllSync creates dir (under root) with 0700, fsyncing the parent of
73// every directory it creates so the new entries are durable.
74func mkdirAllSync(root, dir string) error {
75 var missing []string
76 for d := filepath.Clean(dir); d != filepath.Clean(root) && d != filepath.Dir(d); d = filepath.Dir(d) {
77 if fi, e := os.Stat(d); e == nil {
78 if !fi.IsDir() {
79 return fmt.Errorf("%s: not a directory", d)
80 }
81 break
82 } else if !errors.Is(e, fs.ErrNotExist) {
83 return e
84 }
85 missing = append(missing, d)
86 }
87 for i := len(missing) - 1; i >= 0; i-- {
88 if e := os.Mkdir(missing[i], 0700); e != nil && !errors.Is(e, fs.ErrExist) {
89 return e
90 }
91 if e := syncDir(filepath.Dir(missing[i])); e != nil {
92 return e
93 }
94 }
95 return nil
96}
97
98// reclaimTmp removes entries (files or directories) in root/tmp whose mtime
99// is older than olderThan.
100func (s *Store) reclaimTmp(ctx context.Context, olderThan time.Duration) error {
101 dir := filepath.Join(s.root, "tmp")
102 entries, e := os.ReadDir(dir)
103 if errors.Is(e, fs.ErrNotExist) {
104 return nil
105 }
106 if e != nil {
107 return e
108 }
109 cutoff := now().Add(-olderThan)
110 var errs []error
111 for _, d := range entries {
112 if e = ctx.Err(); e != nil {
113 return e
114 }
115 fi, e := d.Info()
116 if e != nil || !fi.ModTime().Before(cutoff) {
117 continue
118 }
119 if e = os.RemoveAll(filepath.Join(dir, d.Name())); e != nil {
120 errs = append(errs, e)
121 continue
122 }
123 logEntry := zerowrap.FromCtx(ctx)
124 logEntry.Info().Str("entry", d.Name()).Msg("blobdisk: reclaimed stale tmp entry")
125 }
126 return errors.Join(errs...)
127}
128
129type ctxReader struct {
130 ctx context.Context
131 r io.Reader
132}
133
134func (r ctxReader) Read(p []byte) (int, error) {
135 if e := r.ctx.Err(); e != nil {
136 return 0, e
137 }
138 return r.r.Read(p)
139}
140func (s *Store) init(ctx context.Context) error {
141 if e := ctx.Err(); e != nil {
142 return e
143 }
144 for _, d := range []string{"objects", "tmp", "multipart"} {
145 if e := os.MkdirAll(filepath.Join(s.root, d), 0700); e != nil {
146 return e
147 }
148 }
149 return nil
150}
151
152// Check verifies that the root is writable.
153func (s *Store) Check(ctx context.Context) error {
154 logEntry := zerowrap.FromCtx(ctx)
155 logEntry.Debug().Msg("blobdisk: Check")
156 if e := s.init(ctx); e != nil {
157 return e
158 }
159 if e := s.reclaimTmp(ctx, tmpGrace); e != nil {
160 logEntry.Warn().Err(e).Msg("blobdisk: tmp reclamation failed")
161 }
162 f, e := os.CreateTemp(filepath.Join(s.root, "tmp"), ".probe-*")
163 if e != nil {
164 return e
165 }
166 return errors.Join(f.Close(), os.Remove(f.Name()))
167}
168
169// Put atomically replaces key with exactly size bytes.
170func (s *Store) Put(ctx context.Context, key string, r io.Reader, size int64) (ports.BlobInfo, error) {
171 logEntry := zerowrap.FromCtx(ctx)
172 logEntry.Debug().Msg("blobdisk: Put")
173 if e := valid(key); e != nil {
174 return ports.BlobInfo{}, e
175 }
176 if size < 0 {
177 return ports.BlobInfo{}, domain.ErrInvalid
178 }
179 if e := s.init(ctx); e != nil {
180 return ports.BlobInfo{}, e
181 }
182 f, e := os.CreateTemp(filepath.Join(s.root, "tmp"), ".tmp-*")
183 if e != nil {
184 return ports.BlobInfo{}, e
185 }
186 defer os.Remove(f.Name())
187 h := sha256.New()
188 n, e := io.Copy(io.MultiWriter(f, h), ctxReader{ctx, r})
189 if e == nil && n != size {
190 e = domain.ErrInvalid
191 }
192 if e == nil {
193 e = ctx.Err()
194 }
195 if e == nil {
196 e = f.Sync()
197 }
198 e = errors.Join(e, f.Close())
199 if e != nil {
200 return ports.BlobInfo{}, e
201 }
202 path := s.object(key)
203 if e = mkdirAllSync(filepath.Join(s.root, "objects"), filepath.Dir(path)); e == nil {
204 e = os.Rename(f.Name(), path)
205 }
206 if e == nil {
207 e = syncDir(filepath.Dir(path))
208 }
209 if e != nil {
210 return ports.BlobInfo{}, e
211 }
212 info, e := s.Stat(ctx, key)
213 info.Digest = "sha256:" + hex.EncodeToString(h.Sum(nil))
214 return info, e
215}
216
217// Stat returns object metadata.
218func (s *Store) Stat(ctx context.Context, key string) (ports.BlobInfo, error) {
219 if e := valid(key); e != nil {
220 return ports.BlobInfo{}, e
221 }
222 if e := ctx.Err(); e != nil {
223 return ports.BlobInfo{}, e
224 }
225 f, e := os.Stat(s.object(key))
226 if e != nil {
227 return ports.BlobInfo{}, translate(e)
228 }
229 if !f.Mode().IsRegular() {
230 return ports.BlobInfo{}, domain.ErrNotFound
231 }
232 return ports.BlobInfo{Key: key, Size: f.Size(), ModTime: f.ModTime()}, nil
233}
234
235type limitedCloser struct {
236 io.Reader
237 io.Closer
238}
239
240// Open reads an object range.
241//
242//hexcheck:ignore suspicious-business-logic-in-adapter translates the port offset/length contract into a file seek; range rules belong to the port
243func (s *Store) Open(ctx context.Context, key string, offset, length int64) (io.ReadCloser, ports.BlobInfo, error) {
244 info, e := s.Stat(ctx, key)
245 if e != nil {
246 return nil, info, e
247 }
248 if offset < 0 {
249 return nil, info, domain.ErrInvalid
250 }
251 if offset >= info.Size {
252 return nil, info, domain.ErrOutOfRange
253 }
254 f, e := os.Open(s.object(key))
255 if e != nil {
256 return nil, info, translate(e)
257 }
258 if _, e = f.Seek(offset, io.SeekStart); e != nil {
259 f.Close()
260 return nil, info, e
261 }
262 var r io.Reader = ctxReader{ctx, f}
263 if length >= 0 {
264 r = io.LimitReader(r, length)
265 }
266 return limitedCloser{r, f}, info, nil
267}
268
269// Delete removes an object, treating absence as success.
270func (s *Store) Delete(ctx context.Context, key string) error {
271 logEntry := zerowrap.FromCtx(ctx)
272 logEntry.Debug().Msg("blobdisk: Delete")
273 if e := valid(key); e != nil {
274 return e
275 }
276 if e := ctx.Err(); e != nil {
277 return e
278 }
279 e := os.Remove(s.object(key))
280 if errors.Is(e, fs.ErrNotExist) {
281 return nil
282 }
283 return e
284}
285
286// List visits matching objects in lexical order.
287func (s *Store) List(ctx context.Context, prefix string, fn func(ports.BlobInfo) error) error {
288 logEntry := zerowrap.FromCtx(ctx)
289 logEntry.Debug().Msg("blobdisk: List")
290 root := filepath.Join(s.root, "objects")
291 var infos []ports.BlobInfo
292 e := filepath.WalkDir(root, func(path string, d fs.DirEntry, e error) error {
293 if e != nil {
294 return e
295 }
296 if e = ctx.Err(); e != nil {
297 return e
298 }
299 if !d.Type().IsRegular() {
300 return nil
301 }
302 key, e := filepath.Rel(root, path)
303 if e != nil {
304 return e
305 }
306 key = filepath.ToSlash(key)
307 if !strings.HasPrefix(key, prefix) {
308 return nil
309 }
310 i, e := d.Info()
311 if e != nil {
312 return e
313 }
314 infos = append(infos, ports.BlobInfo{Key: key, Size: i.Size(), ModTime: i.ModTime()})
315 return nil
316 })
317 if errors.Is(e, fs.ErrNotExist) {
318 return nil
319 }
320 if e != nil {
321 return e
322 }
323 sort.Slice(infos, func(i, j int) bool { return infos[i].Key < infos[j].Key })
324 for _, i := range infos {
325 if e = ctx.Err(); e != nil {
326 return e
327 }
328 if e = fn(i); e != nil {
329 return e
330 }
331 }
332 return nil
333}
334
335type upload struct {
336 s *Store
337 key, id string
338 done bool
339}
340type metadata struct {
341 Key string
342 Started time.Time
343}
344
345func (u *upload) dir() string { return filepath.Join(u.s.root, "multipart", u.id) }
346
347// NewMultipart starts an isolated staging upload.
348func (s *Store) NewMultipart(ctx context.Context, key string) (ports.MultipartUpload, error) {
349 logEntry := zerowrap.FromCtx(ctx)
350 logEntry.Debug().Msg("blobdisk: NewMultipart")
351 if e := valid(key); e != nil {
352 return nil, e
353 }
354 if e := s.init(ctx); e != nil {
355 return nil, e
356 }
357 b := make([]byte, 16)
358 if _, e := rand.Read(b); e != nil {
359 return nil, e
360 }
361 u := &upload{s: s, key: key, id: hex.EncodeToString(b)}
362 stage := filepath.Join(s.root, "tmp", "mp-"+u.id)
363 if e := os.Mkdir(stage, 0700); e != nil {
364 return nil, e
365 }
366 if e := writeMetadata(stage, metadata{key, time.Now().UTC()}); e != nil {
367 os.RemoveAll(stage)
368 return nil, e
369 }
370 if e := os.Rename(stage, u.dir()); e != nil {
371 os.RemoveAll(stage)
372 return nil, e
373 }
374 if e := syncDir(filepath.Join(s.root, "multipart")); e != nil {
375 os.RemoveAll(u.dir())
376 return nil, e
377 }
378 return u, nil
379}
380
381// writeMetadata durably writes dir/metadata via temp file and rename.
382func writeMetadata(dir string, m metadata) error {
383 data, e := json.Marshal(m)
384 if e != nil {
385 return e
386 }
387 f, e := os.CreateTemp(dir, ".meta-*")
388 if e != nil {
389 return e
390 }
391 _, e = f.Write(data)
392 if e == nil {
393 e = f.Sync()
394 }
395 if e = errors.Join(e, f.Close()); e == nil {
396 e = os.Rename(f.Name(), filepath.Join(dir, "metadata"))
397 }
398 if e != nil {
399 os.Remove(f.Name())
400 return e
401 }
402 return syncDir(dir)
403}
404func (u *upload) ID() string { return u.id }
405func (u *upload) UploadPart(ctx context.Context, n int, r io.Reader, size int64) error {
406 logEntry := zerowrap.FromCtx(ctx)
407 logEntry.Debug().Msg("blobdisk: UploadPart")
408 if u.done || n < 1 || n > ports.BlobMaxParts || size < 0 {
409 return domain.ErrInvalid
410 }
411 f, e := os.CreateTemp(u.dir(), ".tmp-*")
412 if e != nil {
413 return e
414 }
415 defer os.Remove(f.Name())
416 count, e := io.Copy(f, ctxReader{ctx, r})
417 if e == nil && count != size {
418 e = domain.ErrInvalid
419 }
420 if e == nil {
421 e = f.Sync()
422 }
423 e = errors.Join(e, f.Close())
424 if e != nil {
425 return e
426 }
427 return os.Rename(f.Name(), filepath.Join(u.dir(), fmt.Sprintf("%05d", n)))
428}
429func (u *upload) Complete(ctx context.Context) (ports.BlobInfo, error) {
430 logEntry := zerowrap.FromCtx(ctx)
431 logEntry.Debug().Msg("blobdisk: Complete")
432 if u.done {
433 return ports.BlobInfo{}, domain.ErrInvalid
434 }
435 u.done = true
436 defer os.RemoveAll(u.dir())
437 entries, e := os.ReadDir(u.dir())
438 if e != nil {
439 return ports.BlobInfo{}, e
440 }
441 var names []string
442 for _, d := range entries {
443 if n, e := strconv.Atoi(d.Name()); e == nil && n > 0 {
444 names = append(names, d.Name())
445 }
446 }
447 sort.Strings(names)
448 if len(names) == 0 {
449 return ports.BlobInfo{}, domain.ErrInvalid
450 }
451 var size int64
452 var readers []io.Reader
453 var files []*os.File
454 defer func() {
455 for _, f := range files {
456 f.Close()
457 }
458 }()
459 for i, n := range names {
460 f, e := os.Open(filepath.Join(u.dir(), n))
461 if e != nil {
462 return ports.BlobInfo{}, e
463 }
464 files = append(files, f)
465 info, e := f.Stat()
466 if e != nil {
467 return ports.BlobInfo{}, e
468 }
469 if i < len(names)-1 && info.Size() < ports.BlobMinPartSize {
470 return ports.BlobInfo{}, domain.ErrInvalid
471 }
472 size += info.Size()
473 readers = append(readers, f)
474 }
475 info, e := u.s.Put(ctx, u.key, io.MultiReader(readers...), size)
476 info.Digest = ""
477 return info, e
478}
479func (u *upload) Abort(ctx context.Context) error {
480 logEntry := zerowrap.FromCtx(ctx)
481 logEntry.Debug().Msg("blobdisk: Abort")
482 if u.done {
483 return domain.ErrInvalid
484 }
485 u.done = true
486 return u.s.AbortMultipart(ctx, u.key, u.id)
487}
488func validID(id string) bool {
489 b, e := hex.DecodeString(id)
490 return e == nil && len(b) == 16 && id == strings.ToLower(id)
491}
492
493// AbortMultipart removes a listed upload by identity.
494func (s *Store) AbortMultipart(ctx context.Context, key, id string) error {
495 logEntry := zerowrap.FromCtx(ctx)
496 logEntry.Debug().Msg("blobdisk: AbortMultipart")
497 if e := valid(key); e != nil {
498 return e
499 }
500 if !validID(id) {
501 return domain.ErrInvalid
502 }
503 if e := ctx.Err(); e != nil {
504 return e
505 }
506 dir := filepath.Join(s.root, "multipart", id)
507 b, e := os.ReadFile(filepath.Join(dir, "metadata"))
508 if errors.Is(e, fs.ErrNotExist) {
509 return nil
510 }
511 if e != nil {
512 return e
513 }
514 var m metadata
515 if e = json.Unmarshal(b, &m); e != nil {
516 return e
517 }
518 if m.Key != key {
519 return domain.ErrInvalid
520 }
521 return os.RemoveAll(dir)
522}
523
524// ListMultipart visits unfinished uploads matching prefix.
525func (s *Store) ListMultipart(ctx context.Context, prefix string, fn func(ports.MultipartInfo) error) error {
526 logEntry := zerowrap.FromCtx(ctx)
527 logEntry.Debug().Msg("blobdisk: ListMultipart")
528 entries, e := os.ReadDir(filepath.Join(s.root, "multipart"))
529 if errors.Is(e, fs.ErrNotExist) {
530 return nil
531 }
532 if e != nil {
533 return e
534 }
535 if e = s.reclaimTmp(ctx, tmpGrace); e != nil {
536 logEntry.Warn().Err(e).Msg("blobdisk: tmp reclamation failed")
537 }
538 for _, d := range entries {
539 if e = ctx.Err(); e != nil {
540 return e
541 }
542 if !d.IsDir() || !validID(d.Name()) {
543 continue
544 }
545 dir := filepath.Join(s.root, "multipart", d.Name())
546 var m metadata
547 b, e := os.ReadFile(filepath.Join(dir, "metadata"))
548 if e == nil {
549 e = json.Unmarshal(b, &m)
550 }
551 if e == nil && !ports.ValidBlobKey(m.Key) {
552 e = domain.ErrInvalid
553 }
554 if e != nil {
555 logEntry.Warn().Err(e).Str("id", d.Name()).Msg("blobdisk: skipping malformed multipart upload")
556 if fi, ie := d.Info(); ie == nil && fi.ModTime().Before(now().Add(-tmpGrace)) {
557 if re := os.RemoveAll(dir); re != nil {
558 logEntry.Warn().Err(re).Str("id", d.Name()).Msg("blobdisk: remove malformed multipart failed")
559 } else {
560 logEntry.Info().Str("id", d.Name()).Msg("blobdisk: removed malformed multipart upload")
561 }
562 }
563 continue
564 }
565 if strings.HasPrefix(m.Key, prefix) {
566 if e = fn(ports.MultipartInfo{Key: m.Key, ID: d.Name(), Started: m.Started}); e != nil {
567 return e
568 }
569 }
570 }
571 return nil
572}