| 1 | package blob |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "database/sql" |
| 7 | "encoding/hex" |
| 8 | "errors" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "testing" |
| 13 | ) |
| 14 | |
| 15 | func scopeTable(t *testing.T, service *Service) { |
| 16 | t.Helper() |
| 17 | if _, err := service.store.DB().Exec(`CREATE TABLE attachment_scope (sha256 TEXT PRIMARY KEY)` ); err != nil { |
| 18 | t.Fatal(err) |
| 19 | } |
| 20 | } |
| 21 | |
| 22 | func scopeCommit(fail bool, seen *[]bool) func(context.Context, *sql.Tx, Blob, bool) error { |
| 23 | return func(ctx context.Context, tx *sql.Tx, entry Blob, created bool) error { |
| 24 | *seen = append(*seen, created) |
| 25 | if fail { |
| 26 | return errors.New("scope rejected" ) |
| 27 | } |
| 28 | _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO attachment_scope(sha256) VALUES(?)" , entry.SHA256) |
| 29 | return err |
| 30 | } |
| 31 | } |
| 32 | |
| 33 | func TestScopedCommitNewIsAtomic(t *testing.T) { |
| 34 | service := testService(t) |
| 35 | scopeTable(t, service) |
| 36 | body := []byte("atomic scoped upload" ) |
| 37 | digest := sha256.Sum256(body) |
| 38 | hash := hex.EncodeToString(digest[:]) |
| 39 | var seen []bool |
| 40 | _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader(string(body)), Type: "text/plain" , Uploader: testUploader, Commit: scopeCommit(true, &seen)}) |
| 41 | if err == nil || !strings.Contains(err.Error(), "scope rejected" ) { |
| 42 | t.Fatalf("failed scope commit error = %v" , err) |
| 43 | } |
| 44 | if len(seen) != 1 || !seen[0] { |
| 45 | t.Fatalf("commit states = %v, want [true]" , seen) |
| 46 | } |
| 47 | var count int |
| 48 | if err := service.store.DB().QueryRow("SELECT COUNT(*) FROM blobs WHERE sha256=?" , hash).Scan(&count); err != nil { |
| 49 | t.Fatal(err) |
| 50 | } |
| 51 | if count != 0 { |
| 52 | t.Fatalf("rolled back blob metadata count = %d" , count) |
| 53 | } |
| 54 | if _, err := os.Stat(filepath.Join(service.root, hash)); !errors.Is(err, os.ErrNotExist) { |
| 55 | t.Fatalf("rolled back final file stat = %v" , err) |
| 56 | } |
| 57 | } |
| 58 | |
| 59 | func TestScopedCommitDedupAndRecovery(t *testing.T) { |
| 60 | service := testService(t) |
| 61 | scopeTable(t, service) |
| 62 | body := []byte("deduplicated scoped upload" ) |
| 63 | var first []bool |
| 64 | entry, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader(string(body)), Type: "text/plain" , Uploader: testUploader, Commit: scopeCommit(false, &first)}) |
| 65 | if err != nil || len(first) != 1 || !first[0] { |
| 66 | t.Fatalf("initial scoped upload entry=%+v states=%v err=%v" , entry, first, err) |
| 67 | } |
| 68 | other := strings.Repeat("b" , 64) |
| 69 | var rejected []bool |
| 70 | if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader(string(body)), Type: "text/plain" , Uploader: other, Commit: scopeCommit(true, &rejected)}); err == nil { |
| 71 | t.Fatal("rejected deduplicated scope commit succeeded" ) |
| 72 | } |
| 73 | var claims int |
| 74 | if err := service.store.DB().QueryRow("SELECT COUNT(*) FROM blob_claims WHERE sha256=? AND uploader=?" , entry.SHA256, other).Scan(&claims); err != nil { |
| 75 | t.Fatal(err) |
| 76 | } |
| 77 | if claims != 0 { |
| 78 | t.Fatalf("rolled back deduplicated claim count = %d" , claims) |
| 79 | } |
| 80 | var accepted []bool |
| 81 | if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader(string(body)), Type: "text/plain" , Uploader: other, Commit: scopeCommit(false, &accepted)}); err != nil || len(accepted) != 1 || accepted[0] { |
| 82 | t.Fatalf("deduplicated scoped upload states=%v err=%v" , accepted, err) |
| 83 | } |
| 84 | if err := os.Remove(filepath.Join(service.root, entry.SHA256)); err != nil { |
| 85 | t.Fatal(err) |
| 86 | } |
| 87 | if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader(string(body)), Type: "text/plain" , Uploader: testUploader}); err != nil { |
| 88 | t.Fatalf("recovery upload: %v" , err) |
| 89 | } |
| 90 | if _, err := os.Stat(filepath.Join(service.root, entry.SHA256)); err != nil { |
| 91 | t.Fatalf("recovered final file: %v" , err) |
| 92 | } |
| 93 | } |
| 94 | |
| 95 | func TestReconcileStagedBlobRequiresCommittedMetadata(t *testing.T) { |
| 96 | service := testService(t) |
| 97 | body := []byte("staged durable bytes" ) |
| 98 | digest := sha256.Sum256(body) |
| 99 | hash := hex.EncodeToString(digest[:]) |
| 100 | stage := filepath.Join(service.root, ".scoped-" +hash) |
| 101 | if err := os.WriteFile(stage, body, 0o600); err != nil { |
| 102 | t.Fatal(err) |
| 103 | } |
| 104 | if err := reconcile(context.Background(), service.store, service.root); err != nil { |
| 105 | t.Fatal(err) |
| 106 | } |
| 107 | if _, err := os.Stat(stage); !errors.Is(err, os.ErrNotExist) { |
| 108 | t.Fatalf("uncommitted stage stat = %v" , err) |
| 109 | } |
| 110 | if _, err := os.Stat(filepath.Join(service.root, hash)); !errors.Is(err, os.ErrNotExist) { |
| 111 | t.Fatalf("uncommitted final stat = %v" , err) |
| 112 | } |
| 113 | if err := os.WriteFile(stage, body, 0o600); err != nil { |
| 114 | t.Fatal(err) |
| 115 | } |
| 116 | if _, err := service.store.DB().Exec("INSERT INTO blobs(sha256,size,type,uploader,uploaded) VALUES(?,?,?,?,?)" , hash, len(body), "text/plain" , "" , 1); err != nil { |
| 117 | t.Fatal(err) |
| 118 | } |
| 119 | if err := reconcile(context.Background(), service.store, service.root); err != nil { |
| 120 | t.Fatal(err) |
| 121 | } |
| 122 | if _, err := os.Stat(filepath.Join(service.root, hash)); err != nil { |
| 123 | t.Fatalf("committed final stat = %v" , err) |
| 124 | } |
| 125 | } |