fix(blobs): reclaim crashed temp files, skip broken multipart entries, sync new directories
- commit
- 97e2b3377aa02f3c06794606e33f9f6de65456fa
- author
- brice · · 2026-10-21 16:28 UTC
- committer
- brice · · 2026-10-21 17:53 UTC
- parent
- 8361f60
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +214 -11
M internal/adapters/blobdisk/blobdisk.go +127 -11
| @@ -52,7 +52,16 @@ func translate(err error) error { | |||
| 52 | 52 | } | |
| 53 | 53 | return err | |
| 54 | 54 | } | |
| 55 | - | func syncDir(path string) error { | |
| 55 | + | ||
| 56 | + | // tmpGrace is how long leftovers in root/tmp (and malformed multipart dirs) | |
| 57 | + | // are kept before reclamation. | |
| 58 | + | const tmpGrace = time.Hour | |
| 59 | + | ||
| 60 | + | // now is the clock used for reclamation; tests may replace it. | |
| 61 | + | var now = time.Now | |
| 62 | + | ||
| 63 | + | // syncDir fsyncs a directory; tests may replace it to record calls. | |
| 64 | + | var syncDir = func(path string) error { | |
| 56 | 65 | f, e := os.Open(path) | |
| 57 | 66 | if e != nil { | |
| 58 | 67 | return e | |
| @@ -60,6 +69,63 @@ func syncDir(path string) error { | |||
| 60 | 69 | return errors.Join(f.Sync(), f.Close()) | |
| 61 | 70 | } | |
| 62 | 71 | ||
| 72 | + | // mkdirAllSync creates dir (under root) with 0700, fsyncing the parent of | |
| 73 | + | // every directory it creates so the new entries are durable. | |
| 74 | + | func 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. | |
| 100 | + | func (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 | + | ||
| 63 | 129 | type ctxReader struct { | |
| 64 | 130 | ctx context.Context | |
| 65 | 131 | r io.Reader | |
| @@ -90,6 +156,9 @@ func (s *Store) Check(ctx context.Context) error { | |||
| 90 | 156 | if e := s.init(ctx); e != nil { | |
| 91 | 157 | return e | |
| 92 | 158 | } | |
| 159 | + | if e := s.reclaimTmp(ctx, tmpGrace); e != nil { | |
| 160 | + | logEntry.Warn().Err(e).Msg("blobdisk: tmp reclamation failed") | |
| 161 | + | } | |
| 93 | 162 | f, e := os.CreateTemp(filepath.Join(s.root, "tmp"), ".probe-*") | |
| 94 | 163 | if e != nil { | |
| 95 | 164 | return e | |
| @@ -131,7 +200,7 @@ func (s *Store) Put(ctx context.Context, key string, r io.Reader, size int64) (p | |||
| 131 | 200 | return ports.BlobInfo{}, e | |
| 132 | 201 | } | |
| 133 | 202 | path := s.object(key) | |
| 134 | - | if e = os.MkdirAll(filepath.Dir(path), 0700); e == nil { | |
| 203 | + | if e = mkdirAllSync(filepath.Join(s.root, "objects"), filepath.Dir(path)); e == nil { | |
| 135 | 204 | e = os.Rename(f.Name(), path) | |
| 136 | 205 | } | |
| 137 | 206 | if e == nil { | |
| @@ -290,16 +359,48 @@ func (s *Store) NewMultipart(ctx context.Context, key string) (ports.MultipartUp | |||
| 290 | 359 | return nil, e | |
| 291 | 360 | } | |
| 292 | 361 | u := &upload{s: s, key: key, id: hex.EncodeToString(b)} | |
| 293 | - | if e := os.Mkdir(u.dir(), 0700); e != nil { | |
| 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) | |
| 294 | 368 | return nil, e | |
| 295 | 369 | } | |
| 296 | - | data, _ := json.Marshal(metadata{key, time.Now().UTC()}) | |
| 297 | - | if e := os.WriteFile(filepath.Join(u.dir(), "metadata"), data, 0600); e != nil { | |
| 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 { | |
| 298 | 375 | os.RemoveAll(u.dir()) | |
| 299 | 376 | return nil, e | |
| 300 | 377 | } | |
| 301 | 378 | return u, nil | |
| 302 | 379 | } | |
| 380 | + | ||
| 381 | + | // writeMetadata durably writes dir/metadata via temp file and rename. | |
| 382 | + | func 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 | + | } | |
| 303 | 404 | func (u *upload) ID() string { return u.id } | |
| 304 | 405 | func (u *upload) UploadPart(ctx context.Context, n int, r io.Reader, size int64) error { | |
| 305 | 406 | logEntry := zerowrap.FromCtx(ctx) | |
| @@ -431,6 +532,9 @@ func (s *Store) ListMultipart(ctx context.Context, prefix string, fn func(ports. | |||
| 431 | 532 | if e != nil { | |
| 432 | 533 | return e | |
| 433 | 534 | } | |
| 535 | + | if e = s.reclaimTmp(ctx, tmpGrace); e != nil { | |
| 536 | + | logEntry.Warn().Err(e).Msg("blobdisk: tmp reclamation failed") | |
| 537 | + | } | |
| 434 | 538 | for _, d := range entries { | |
| 435 | 539 | if e = ctx.Err(); e != nil { | |
| 436 | 540 | return e | |
| @@ -438,13 +542,25 @@ func (s *Store) ListMultipart(ctx context.Context, prefix string, fn func(ports. | |||
| 438 | 542 | if !d.IsDir() || !validID(d.Name()) { | |
| 439 | 543 | continue | |
| 440 | 544 | } | |
| 441 | - | b, e := os.ReadFile(filepath.Join(s.root, "multipart", d.Name(), "metadata")) | |
| 442 | - | if e != nil { | |
| 443 | - | return e | |
| 444 | - | } | |
| 545 | + | dir := filepath.Join(s.root, "multipart", d.Name()) | |
| 445 | 546 | var m metadata | |
| 446 | - | if e = json.Unmarshal(b, &m); e != nil { | |
| 447 | - | return e | |
| 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 | |
| 448 | 564 | } | |
| 449 | 565 | if strings.HasPrefix(m.Key, prefix) { | |
| 450 | 566 | if e = fn(ports.MultipartInfo{Key: m.Key, ID: d.Name(), Started: m.Started}); e != nil { | |
| @@ -52,7 +52,16 @@ func translate(err error) error { | |||||
| 52 | } | 52 | } | ||
| 53 | return err | 53 | return err | ||
| 54 | } | 54 | } | ||
| 55 | - | func syncDir(path string) error { | 55 | + | |
| 56 | + | // tmpGrace is how long leftovers in root/tmp (and malformed multipart dirs) | |||
| 57 | + | // are kept before reclamation. | |||
| 58 | + | const tmpGrace = time.Hour | |||
| 59 | + | ||||
| 60 | + | // now is the clock used for reclamation; tests may replace it. | |||
| 61 | + | var now = time.Now | |||
| 62 | + | ||||
| 63 | + | // syncDir fsyncs a directory; tests may replace it to record calls. | |||
| 64 | + | var syncDir = func(path string) error { | |||
| 56 | f, e := os.Open(path) | 65 | f, e := os.Open(path) | ||
| 57 | if e != nil { | 66 | if e != nil { | ||
| 58 | return e | 67 | return e | ||
| @@ -60,6 +69,63 @@ func syncDir(path string) error { | |||||
| 60 | return errors.Join(f.Sync(), f.Close()) | 69 | return errors.Join(f.Sync(), f.Close()) | ||
| 61 | } | 70 | } | ||
| 62 | 71 | ||||
| 72 | + | // mkdirAllSync creates dir (under root) with 0700, fsyncing the parent of | |||
| 73 | + | // every directory it creates so the new entries are durable. | |||
| 74 | + | func 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. | |||
| 100 | + | func (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 | + | ||||
| 63 | type ctxReader struct { | 129 | type ctxReader struct { | ||
| 64 | ctx context.Context | 130 | ctx context.Context | ||
| 65 | r io.Reader | 131 | r io.Reader | ||
| @@ -90,6 +156,9 @@ func (s *Store) Check(ctx context.Context) error { | |||||
| 90 | if e := s.init(ctx); e != nil { | 156 | if e := s.init(ctx); e != nil { | ||
| 91 | return e | 157 | return e | ||
| 92 | } | 158 | } | ||
| 159 | + | if e := s.reclaimTmp(ctx, tmpGrace); e != nil { | |||
| 160 | + | logEntry.Warn().Err(e).Msg("blobdisk: tmp reclamation failed") | |||
| 161 | + | } | |||
| 93 | f, e := os.CreateTemp(filepath.Join(s.root, "tmp"), ".probe-*") | 162 | f, e := os.CreateTemp(filepath.Join(s.root, "tmp"), ".probe-*") | ||
| 94 | if e != nil { | 163 | if e != nil { | ||
| 95 | return e | 164 | return e | ||
| @@ -131,7 +200,7 @@ func (s *Store) Put(ctx context.Context, key string, r io.Reader, size int64) (p | |||||
| 131 | return ports.BlobInfo{}, e | 200 | return ports.BlobInfo{}, e | ||
| 132 | } | 201 | } | ||
| 133 | path := s.object(key) | 202 | path := s.object(key) | ||
| 134 | - | if e = os.MkdirAll(filepath.Dir(path), 0700); e == nil { | 203 | + | if e = mkdirAllSync(filepath.Join(s.root, "objects"), filepath.Dir(path)); e == nil { |
| 135 | e = os.Rename(f.Name(), path) | 204 | e = os.Rename(f.Name(), path) | ||
| 136 | } | 205 | } | ||
| 137 | if e == nil { | 206 | if e == nil { | ||
| @@ -290,16 +359,48 @@ func (s *Store) NewMultipart(ctx context.Context, key string) (ports.MultipartUp | |||||
| 290 | return nil, e | 359 | return nil, e | ||
| 291 | } | 360 | } | ||
| 292 | u := &upload{s: s, key: key, id: hex.EncodeToString(b)} | 361 | u := &upload{s: s, key: key, id: hex.EncodeToString(b)} | ||
| 293 | - | if e := os.Mkdir(u.dir(), 0700); e != nil { | 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) | |||
| 294 | return nil, e | 368 | return nil, e | ||
| 295 | } | 369 | } | ||
| 296 | - | data, _ := json.Marshal(metadata{key, time.Now().UTC()}) | 370 | + | if e := os.Rename(stage, u.dir()); e != nil { |
| 297 | - | if e := os.WriteFile(filepath.Join(u.dir(), "metadata"), data, 0600); e != nil { | 371 | + | os.RemoveAll(stage) |
| 372 | + | return nil, e | |||
| 373 | + | } | |||
| 374 | + | if e := syncDir(filepath.Join(s.root, "multipart")); e != nil { | |||
| 298 | os.RemoveAll(u.dir()) | 375 | os.RemoveAll(u.dir()) | ||
| 299 | return nil, e | 376 | return nil, e | ||
| 300 | } | 377 | } | ||
| 301 | return u, nil | 378 | return u, nil | ||
| 302 | } | 379 | } | ||
| 380 | + | ||||
| 381 | + | // writeMetadata durably writes dir/metadata via temp file and rename. | |||
| 382 | + | func 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 | + | } | |||
| 303 | func (u *upload) ID() string { return u.id } | 404 | func (u *upload) ID() string { return u.id } | ||
| 304 | func (u *upload) UploadPart(ctx context.Context, n int, r io.Reader, size int64) error { | 405 | func (u *upload) UploadPart(ctx context.Context, n int, r io.Reader, size int64) error { | ||
| 305 | logEntry := zerowrap.FromCtx(ctx) | 406 | logEntry := zerowrap.FromCtx(ctx) | ||
| @@ -431,6 +532,9 @@ func (s *Store) ListMultipart(ctx context.Context, prefix string, fn func(ports. | |||||
| 431 | if e != nil { | 532 | if e != nil { | ||
| 432 | return e | 533 | return e | ||
| 433 | } | 534 | } | ||
| 535 | + | if e = s.reclaimTmp(ctx, tmpGrace); e != nil { | |||
| 536 | + | logEntry.Warn().Err(e).Msg("blobdisk: tmp reclamation failed") | |||
| 537 | + | } | |||
| 434 | for _, d := range entries { | 538 | for _, d := range entries { | ||
| 435 | if e = ctx.Err(); e != nil { | 539 | if e = ctx.Err(); e != nil { | ||
| 436 | return e | 540 | return e | ||
| @@ -438,13 +542,25 @@ func (s *Store) ListMultipart(ctx context.Context, prefix string, fn func(ports. | |||||
| 438 | if !d.IsDir() || !validID(d.Name()) { | 542 | if !d.IsDir() || !validID(d.Name()) { | ||
| 439 | continue | 543 | continue | ||
| 440 | } | 544 | } | ||
| 441 | - | b, e := os.ReadFile(filepath.Join(s.root, "multipart", d.Name(), "metadata")) | 545 | + | dir := filepath.Join(s.root, "multipart", d.Name()) |
| 442 | - | if e != nil { | |||
| 443 | - | return e | |||
| 444 | - | } | |||
| 445 | var m metadata | 546 | var m metadata | ||
| 446 | - | if e = json.Unmarshal(b, &m); e != nil { | 547 | + | b, e := os.ReadFile(filepath.Join(dir, "metadata")) |
| 447 | - | return e | 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 | |||
| 448 | } | 564 | } | ||
| 449 | if strings.HasPrefix(m.Key, prefix) { | 565 | if strings.HasPrefix(m.Key, prefix) { | ||
| 450 | if e = fn(ports.MultipartInfo{Key: m.Key, ID: d.Name(), Started: m.Started}); e != nil { | 566 | if e = fn(ports.MultipartInfo{Key: m.Key, ID: d.Name(), Started: m.Started}); e != nil { | ||
A internal/adapters/blobdisk/blobdisk_internal_test.go +87 -0
| @@ -0,0 +1,87 @@ | |||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |
| 2 | + | package blobdisk | |
| 3 | + | ||
| 4 | + | import ( | |
| 5 | + | "context" | |
| 6 | + | "os" | |
| 7 | + | "path/filepath" | |
| 8 | + | "strings" | |
| 9 | + | "testing" | |
| 10 | + | "time" | |
| 11 | + | ||
| 12 | + | "bnema.dev/ermite/internal/ports" | |
| 13 | + | "github.com/stretchr/testify/require" | |
| 14 | + | ) | |
| 15 | + | ||
| 16 | + | func age(t *testing.T, p string) { | |
| 17 | + | t.Helper() | |
| 18 | + | old := time.Now().Add(-2 * tmpGrace) | |
| 19 | + | require.NoError(t, os.Chtimes(p, old, old)) | |
| 20 | + | } | |
| 21 | + | ||
| 22 | + | func TestCheckReclaimsOldTmp(t *testing.T) { | |
| 23 | + | root := t.TempDir() | |
| 24 | + | s := New(root) | |
| 25 | + | ctx := context.Background() | |
| 26 | + | require.NoError(t, s.Check(ctx)) | |
| 27 | + | oldF, freshF := filepath.Join(root, "tmp", ".tmp-old"), filepath.Join(root, "tmp", ".tmp-fresh") | |
| 28 | + | oldD := filepath.Join(root, "tmp", "mp-old") | |
| 29 | + | require.NoError(t, os.WriteFile(oldF, []byte("x"), 0600)) | |
| 30 | + | require.NoError(t, os.WriteFile(freshF, []byte("x"), 0600)) | |
| 31 | + | require.NoError(t, os.MkdirAll(oldD, 0700)) | |
| 32 | + | require.NoError(t, os.WriteFile(filepath.Join(oldD, "metadata"), []byte("x"), 0600)) | |
| 33 | + | age(t, oldF) | |
| 34 | + | age(t, oldD) | |
| 35 | + | require.NoError(t, s.Check(ctx)) | |
| 36 | + | require.NoFileExists(t, oldF) | |
| 37 | + | require.NoDirExists(t, oldD) | |
| 38 | + | require.FileExists(t, freshF) | |
| 39 | + | } | |
| 40 | + | ||
| 41 | + | func TestListMultipartSkipsAndReclaimsMalformed(t *testing.T) { | |
| 42 | + | root := t.TempDir() | |
| 43 | + | s := New(root) | |
| 44 | + | ctx := context.Background() | |
| 45 | + | a, err := s.NewMultipart(ctx, "a") | |
| 46 | + | require.NoError(t, err) | |
| 47 | + | c, err := s.NewMultipart(ctx, "c") | |
| 48 | + | require.NoError(t, err) | |
| 49 | + | mk := func(id string, meta *string) string { | |
| 50 | + | d := filepath.Join(root, "multipart", id) | |
| 51 | + | require.NoError(t, os.Mkdir(d, 0700)) | |
| 52 | + | if meta != nil { | |
| 53 | + | require.NoError(t, os.WriteFile(filepath.Join(d, "metadata"), []byte(*meta), 0600)) | |
| 54 | + | } | |
| 55 | + | return d | |
| 56 | + | } | |
| 57 | + | trunc := `{"Key":` | |
| 58 | + | freshBad := mk(strings.Repeat("1", 32), &trunc) | |
| 59 | + | oldBad := mk(strings.Repeat("2", 32), nil) | |
| 60 | + | oldTrunc := mk(strings.Repeat("3", 32), &trunc) | |
| 61 | + | age(t, oldBad) | |
| 62 | + | age(t, oldTrunc) | |
| 63 | + | var got []string | |
| 64 | + | require.NoError(t, s.ListMultipart(ctx, "", func(i ports.MultipartInfo) error { got = append(got, i.ID); return nil })) | |
| 65 | + | require.ElementsMatch(t, []string{a.ID(), c.ID()}, got) | |
| 66 | + | require.DirExists(t, freshBad) | |
| 67 | + | require.NoDirExists(t, oldBad) | |
| 68 | + | require.NoDirExists(t, oldTrunc) | |
| 69 | + | entries, err := os.ReadDir(filepath.Join(root, "tmp")) | |
| 70 | + | require.NoError(t, err) | |
| 71 | + | require.Empty(t, entries) | |
| 72 | + | } | |
| 73 | + | ||
| 74 | + | func TestMkdirAllSyncSyncsParents(t *testing.T) { | |
| 75 | + | root := t.TempDir() | |
| 76 | + | s := New(root) | |
| 77 | + | var synced []string | |
| 78 | + | orig := syncDir | |
| 79 | + | syncDir = func(p string) error { synced = append(synced, p); return orig(p) } | |
| 80 | + | t.Cleanup(func() { syncDir = orig }) | |
| 81 | + | _, err := s.Put(context.Background(), "a/b/c/obj", strings.NewReader("hi"), 2) | |
| 82 | + | require.NoError(t, err) | |
| 83 | + | obj := filepath.Join(root, "objects") | |
| 84 | + | for _, want := range []string{obj, filepath.Join(obj, "a"), filepath.Join(obj, "a/b"), filepath.Join(obj, "a/b/c")} { | |
| 85 | + | require.Contains(t, synced, want) | |
| 86 | + | } | |
| 87 | + | } | |
| @@ -0,0 +1,87 @@ | |||||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |||
| 2 | + | package blobdisk | |||
| 3 | + | ||||
| 4 | + | import ( | |||
| 5 | + | "context" | |||
| 6 | + | "os" | |||
| 7 | + | "path/filepath" | |||
| 8 | + | "strings" | |||
| 9 | + | "testing" | |||
| 10 | + | "time" | |||
| 11 | + | ||||
| 12 | + | "bnema.dev/ermite/internal/ports" | |||
| 13 | + | "github.com/stretchr/testify/require" | |||
| 14 | + | ) | |||
| 15 | + | ||||
| 16 | + | func age(t *testing.T, p string) { | |||
| 17 | + | t.Helper() | |||
| 18 | + | old := time.Now().Add(-2 * tmpGrace) | |||
| 19 | + | require.NoError(t, os.Chtimes(p, old, old)) | |||
| 20 | + | } | |||
| 21 | + | ||||
| 22 | + | func TestCheckReclaimsOldTmp(t *testing.T) { | |||
| 23 | + | root := t.TempDir() | |||
| 24 | + | s := New(root) | |||
| 25 | + | ctx := context.Background() | |||
| 26 | + | require.NoError(t, s.Check(ctx)) | |||
| 27 | + | oldF, freshF := filepath.Join(root, "tmp", ".tmp-old"), filepath.Join(root, "tmp", ".tmp-fresh") | |||
| 28 | + | oldD := filepath.Join(root, "tmp", "mp-old") | |||
| 29 | + | require.NoError(t, os.WriteFile(oldF, []byte("x"), 0600)) | |||
| 30 | + | require.NoError(t, os.WriteFile(freshF, []byte("x"), 0600)) | |||
| 31 | + | require.NoError(t, os.MkdirAll(oldD, 0700)) | |||
| 32 | + | require.NoError(t, os.WriteFile(filepath.Join(oldD, "metadata"), []byte("x"), 0600)) | |||
| 33 | + | age(t, oldF) | |||
| 34 | + | age(t, oldD) | |||
| 35 | + | require.NoError(t, s.Check(ctx)) | |||
| 36 | + | require.NoFileExists(t, oldF) | |||
| 37 | + | require.NoDirExists(t, oldD) | |||
| 38 | + | require.FileExists(t, freshF) | |||
| 39 | + | } | |||
| 40 | + | ||||
| 41 | + | func TestListMultipartSkipsAndReclaimsMalformed(t *testing.T) { | |||
| 42 | + | root := t.TempDir() | |||
| 43 | + | s := New(root) | |||
| 44 | + | ctx := context.Background() | |||
| 45 | + | a, err := s.NewMultipart(ctx, "a") | |||
| 46 | + | require.NoError(t, err) | |||
| 47 | + | c, err := s.NewMultipart(ctx, "c") | |||
| 48 | + | require.NoError(t, err) | |||
| 49 | + | mk := func(id string, meta *string) string { | |||
| 50 | + | d := filepath.Join(root, "multipart", id) | |||
| 51 | + | require.NoError(t, os.Mkdir(d, 0700)) | |||
| 52 | + | if meta != nil { | |||
| 53 | + | require.NoError(t, os.WriteFile(filepath.Join(d, "metadata"), []byte(*meta), 0600)) | |||
| 54 | + | } | |||
| 55 | + | return d | |||
| 56 | + | } | |||
| 57 | + | trunc := `{"Key":` | |||
| 58 | + | freshBad := mk(strings.Repeat("1", 32), &trunc) | |||
| 59 | + | oldBad := mk(strings.Repeat("2", 32), nil) | |||
| 60 | + | oldTrunc := mk(strings.Repeat("3", 32), &trunc) | |||
| 61 | + | age(t, oldBad) | |||
| 62 | + | age(t, oldTrunc) | |||
| 63 | + | var got []string | |||
| 64 | + | require.NoError(t, s.ListMultipart(ctx, "", func(i ports.MultipartInfo) error { got = append(got, i.ID); return nil })) | |||
| 65 | + | require.ElementsMatch(t, []string{a.ID(), c.ID()}, got) | |||
| 66 | + | require.DirExists(t, freshBad) | |||
| 67 | + | require.NoDirExists(t, oldBad) | |||
| 68 | + | require.NoDirExists(t, oldTrunc) | |||
| 69 | + | entries, err := os.ReadDir(filepath.Join(root, "tmp")) | |||
| 70 | + | require.NoError(t, err) | |||
| 71 | + | require.Empty(t, entries) | |||
| 72 | + | } | |||
| 73 | + | ||||
| 74 | + | func TestMkdirAllSyncSyncsParents(t *testing.T) { | |||
| 75 | + | root := t.TempDir() | |||
| 76 | + | s := New(root) | |||
| 77 | + | var synced []string | |||
| 78 | + | orig := syncDir | |||
| 79 | + | syncDir = func(p string) error { synced = append(synced, p); return orig(p) } | |||
| 80 | + | t.Cleanup(func() { syncDir = orig }) | |||
| 81 | + | _, err := s.Put(context.Background(), "a/b/c/obj", strings.NewReader("hi"), 2) | |||
| 82 | + | require.NoError(t, err) | |||
| 83 | + | obj := filepath.Join(root, "objects") | |||
| 84 | + | for _, want := range []string{obj, filepath.Join(obj, "a"), filepath.Join(obj, "a/b"), filepath.Join(obj, "a/b/c")} { | |||
| 85 | + | require.Contains(t, synced, want) | |||
| 86 | + | } | |||
| 87 | + | } | |||