feat(kvstore): store repository metadata in ermite-kv
- commit
- 3c97d930635ae2c75921ceacf5270bec8fae3cd6
- author
- brice · · 2026-10-10 06:13 UTC
- parent
- c5c8a63
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +545 -0
A internal/adapters/kvstore/repos.go +173 -0
| @@ -0,0 +1,173 @@ | |||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |
| 2 | + | ||
| 3 | + | package kvstore | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "context" | |
| 7 | + | "errors" | |
| 8 | + | "fmt" | |
| 9 | + | "time" | |
| 10 | + | "unicode/utf8" | |
| 11 | + | ||
| 12 | + | kv "bnema.dev/ermite-kv" | |
| 13 | + | "google.golang.org/protobuf/proto" | |
| 14 | + | "google.golang.org/protobuf/types/known/timestamppb" | |
| 15 | + | ||
| 16 | + | storagepb "bnema.dev/ermite/gen/ermite/storage" | |
| 17 | + | "bnema.dev/ermite/internal/domain" | |
| 18 | + | ) | |
| 19 | + | ||
| 20 | + | const repoPrefix = "repo/" | |
| 21 | + | ||
| 22 | + | func repoKey(name domain.RepoName) []byte { return []byte(repoPrefix + name.String()) } | |
| 23 | + | ||
| 24 | + | // Create stores repo with an "absent" precondition: a repository with that | |
| 25 | + | // name, or a concurrent Create that commits first, yields domain.ErrAlreadyExists. | |
| 26 | + | func (s *Store) Create(ctx context.Context, repo domain.Repo) error { | |
| 27 | + | val, err := encodeRepo(repo) | |
| 28 | + | if err != nil { | |
| 29 | + | return err | |
| 30 | + | } | |
| 31 | + | key := repoKey(repo.Name) | |
| 32 | + | var b kv.Batch | |
| 33 | + | b.RequireAbsent(key) | |
| 34 | + | b.Put(key, val) | |
| 35 | + | if _, err := s.db.Apply(ctx, &b); err != nil { | |
| 36 | + | if errors.Is(err, kv.ErrConflict) { | |
| 37 | + | return domain.ErrAlreadyExists | |
| 38 | + | } | |
| 39 | + | return fmt.Errorf("create repository %q: %w", repo.Name, err) | |
| 40 | + | } | |
| 41 | + | return nil | |
| 42 | + | } | |
| 43 | + | ||
| 44 | + | // Get returns the repository or domain.ErrNotFound. | |
| 45 | + | func (s *Store) Get(ctx context.Context, name domain.RepoName) (domain.Repo, error) { | |
| 46 | + | item, ok := s.db.Get(ctx, repoKey(name)) | |
| 47 | + | if !ok { | |
| 48 | + | return domain.Repo{}, domain.ErrNotFound | |
| 49 | + | } | |
| 50 | + | return decodeRepo(item.Value) | |
| 51 | + | } | |
| 52 | + | ||
| 53 | + | // List returns every repository ordered by name. | |
| 54 | + | func (s *Store) List(ctx context.Context) ([]domain.Repo, error) { | |
| 55 | + | var repos []domain.Repo | |
| 56 | + | for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(repoPrefix)) { | |
| 57 | + | repo, err := decodeRepo(item.Value) | |
| 58 | + | if err != nil { | |
| 59 | + | return nil, fmt.Errorf("decode %s: %w", key, err) | |
| 60 | + | } | |
| 61 | + | repos = append(repos, repo) | |
| 62 | + | } | |
| 63 | + | return repos, nil | |
| 64 | + | } | |
| 65 | + | ||
| 66 | + | // Update replaces the stored repository. It returns domain.ErrNotFound if | |
| 67 | + | // there is none and domain.ErrConflict if another write changed it since it | |
| 68 | + | // was read. | |
| 69 | + | // | |
| 70 | + | //hexcheck:ignore suspicious-business-logic-in-adapter port method name only | |
| 71 | + | func (s *Store) Update(ctx context.Context, repo domain.Repo) error { | |
| 72 | + | val, err := encodeRepo(repo) | |
| 73 | + | if err != nil { | |
| 74 | + | return err | |
| 75 | + | } | |
| 76 | + | key := repoKey(repo.Name) | |
| 77 | + | item, ok := s.db.Get(ctx, key) | |
| 78 | + | if !ok { | |
| 79 | + | return domain.ErrNotFound | |
| 80 | + | } | |
| 81 | + | var b kv.Batch | |
| 82 | + | b.RequireVersion(key, item.Version) | |
| 83 | + | b.Put(key, val) | |
| 84 | + | if _, err := s.db.Apply(ctx, &b); err != nil { | |
| 85 | + | if errors.Is(err, kv.ErrConflict) { | |
| 86 | + | return domain.ErrConflict | |
| 87 | + | } | |
| 88 | + | return fmt.Errorf("update repository %q: %w", repo.Name, err) | |
| 89 | + | } | |
| 90 | + | return nil | |
| 91 | + | } | |
| 92 | + | ||
| 93 | + | // Delete removes the repository; a missing one is not an error. | |
| 94 | + | func (s *Store) Delete(ctx context.Context, name domain.RepoName) error { | |
| 95 | + | var b kv.Batch | |
| 96 | + | b.Delete(repoKey(name)) | |
| 97 | + | if _, err := s.db.Apply(ctx, &b); err != nil { | |
| 98 | + | return fmt.Errorf("delete repository %q: %w", name, err) | |
| 99 | + | } | |
| 100 | + | return nil | |
| 101 | + | } | |
| 102 | + | ||
| 103 | + | func encodeRepo(r domain.Repo) ([]byte, error) { | |
| 104 | + | if r.Name.IsZero() { | |
| 105 | + | return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName) | |
| 106 | + | } | |
| 107 | + | if !utf8.ValidString(r.Description) { | |
| 108 | + | return nil, errors.New("repository description is not valid UTF-8") | |
| 109 | + | } | |
| 110 | + | vis, err := encodeVisibility(r.Visibility) | |
| 111 | + | if err != nil { | |
| 112 | + | return nil, err | |
| 113 | + | } | |
| 114 | + | pb := &storagepb.Repo{ | |
| 115 | + | Name: r.Name.String(), | |
| 116 | + | Description: r.Description, | |
| 117 | + | Visibility: vis, | |
| 118 | + | CreatedAt: encodeTime(r.CreatedAt), | |
| 119 | + | PushedAt: encodeTime(r.PushedAt), | |
| 120 | + | } | |
| 121 | + | return mustMarshal(pb), nil | |
| 122 | + | } | |
| 123 | + | ||
| 124 | + | func decodeRepo(val []byte) (domain.Repo, error) { | |
| 125 | + | var pb storagepb.Repo | |
| 126 | + | if err := proto.Unmarshal(val, &pb); err != nil { | |
| 127 | + | return domain.Repo{}, fmt.Errorf("decode repository: %w", err) | |
| 128 | + | } | |
| 129 | + | name, err := domain.ParseRepoName(pb.GetName()) | |
| 130 | + | if err != nil { | |
| 131 | + | return domain.Repo{}, fmt.Errorf("decode repository: %w", err) | |
| 132 | + | } | |
| 133 | + | return domain.Repo{ | |
| 134 | + | Name: name, | |
| 135 | + | Description: pb.GetDescription(), | |
| 136 | + | Visibility: decodeVisibility(pb.GetVisibility()), | |
| 137 | + | CreatedAt: decodeTime(pb.GetCreatedAt()), | |
| 138 | + | PushedAt: decodeTime(pb.GetPushedAt()), | |
| 139 | + | }, nil | |
| 140 | + | } | |
| 141 | + | ||
| 142 | + | func encodeVisibility(v domain.Visibility) (storagepb.Visibility, error) { | |
| 143 | + | switch v { | |
| 144 | + | case domain.VisibilityPublic: | |
| 145 | + | return storagepb.Visibility_VISIBILITY_PUBLIC, nil | |
| 146 | + | case domain.VisibilityPrivate: | |
| 147 | + | return storagepb.Visibility_VISIBILITY_PRIVATE, nil | |
| 148 | + | } | |
| 149 | + | return storagepb.Visibility_VISIBILITY_UNSPECIFIED, fmt.Errorf("%w: %q", domain.ErrInvalidVisibility, v) | |
| 150 | + | } | |
| 151 | + | ||
| 152 | + | // decodeVisibility maps anything but public to private, including the | |
| 153 | + | // unspecified value and values from a newer schema: the safe side. | |
| 154 | + | func decodeVisibility(v storagepb.Visibility) domain.Visibility { | |
| 155 | + | if v == storagepb.Visibility_VISIBILITY_PUBLIC { | |
| 156 | + | return domain.VisibilityPublic | |
| 157 | + | } | |
| 158 | + | return domain.VisibilityPrivate | |
| 159 | + | } | |
| 160 | + | ||
| 161 | + | func encodeTime(t time.Time) *timestamppb.Timestamp { | |
| 162 | + | if t.IsZero() { | |
| 163 | + | return nil | |
| 164 | + | } | |
| 165 | + | return timestamppb.New(t) | |
| 166 | + | } | |
| 167 | + | ||
| 168 | + | func decodeTime(ts *timestamppb.Timestamp) time.Time { | |
| 169 | + | if ts == nil { | |
| 170 | + | return time.Time{} | |
| 171 | + | } | |
| 172 | + | return ts.AsTime() | |
| 173 | + | } | |
| @@ -0,0 +1,173 @@ | |||||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |||
| 2 | + | ||||
| 3 | + | package kvstore | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "context" | |||
| 7 | + | "errors" | |||
| 8 | + | "fmt" | |||
| 9 | + | "time" | |||
| 10 | + | "unicode/utf8" | |||
| 11 | + | ||||
| 12 | + | kv "bnema.dev/ermite-kv" | |||
| 13 | + | "google.golang.org/protobuf/proto" | |||
| 14 | + | "google.golang.org/protobuf/types/known/timestamppb" | |||
| 15 | + | ||||
| 16 | + | storagepb "bnema.dev/ermite/gen/ermite/storage" | |||
| 17 | + | "bnema.dev/ermite/internal/domain" | |||
| 18 | + | ) | |||
| 19 | + | ||||
| 20 | + | const repoPrefix = "repo/" | |||
| 21 | + | ||||
| 22 | + | func repoKey(name domain.RepoName) []byte { return []byte(repoPrefix + name.String()) } | |||
| 23 | + | ||||
| 24 | + | // Create stores repo with an "absent" precondition: a repository with that | |||
| 25 | + | // name, or a concurrent Create that commits first, yields domain.ErrAlreadyExists. | |||
| 26 | + | func (s *Store) Create(ctx context.Context, repo domain.Repo) error { | |||
| 27 | + | val, err := encodeRepo(repo) | |||
| 28 | + | if err != nil { | |||
| 29 | + | return err | |||
| 30 | + | } | |||
| 31 | + | key := repoKey(repo.Name) | |||
| 32 | + | var b kv.Batch | |||
| 33 | + | b.RequireAbsent(key) | |||
| 34 | + | b.Put(key, val) | |||
| 35 | + | if _, err := s.db.Apply(ctx, &b); err != nil { | |||
| 36 | + | if errors.Is(err, kv.ErrConflict) { | |||
| 37 | + | return domain.ErrAlreadyExists | |||
| 38 | + | } | |||
| 39 | + | return fmt.Errorf("create repository %q: %w", repo.Name, err) | |||
| 40 | + | } | |||
| 41 | + | return nil | |||
| 42 | + | } | |||
| 43 | + | ||||
| 44 | + | // Get returns the repository or domain.ErrNotFound. | |||
| 45 | + | func (s *Store) Get(ctx context.Context, name domain.RepoName) (domain.Repo, error) { | |||
| 46 | + | item, ok := s.db.Get(ctx, repoKey(name)) | |||
| 47 | + | if !ok { | |||
| 48 | + | return domain.Repo{}, domain.ErrNotFound | |||
| 49 | + | } | |||
| 50 | + | return decodeRepo(item.Value) | |||
| 51 | + | } | |||
| 52 | + | ||||
| 53 | + | // List returns every repository ordered by name. | |||
| 54 | + | func (s *Store) List(ctx context.Context) ([]domain.Repo, error) { | |||
| 55 | + | var repos []domain.Repo | |||
| 56 | + | for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(repoPrefix)) { | |||
| 57 | + | repo, err := decodeRepo(item.Value) | |||
| 58 | + | if err != nil { | |||
| 59 | + | return nil, fmt.Errorf("decode %s: %w", key, err) | |||
| 60 | + | } | |||
| 61 | + | repos = append(repos, repo) | |||
| 62 | + | } | |||
| 63 | + | return repos, nil | |||
| 64 | + | } | |||
| 65 | + | ||||
| 66 | + | // Update replaces the stored repository. It returns domain.ErrNotFound if | |||
| 67 | + | // there is none and domain.ErrConflict if another write changed it since it | |||
| 68 | + | // was read. | |||
| 69 | + | // | |||
| 70 | + | //hexcheck:ignore suspicious-business-logic-in-adapter port method name only | |||
| 71 | + | func (s *Store) Update(ctx context.Context, repo domain.Repo) error { | |||
| 72 | + | val, err := encodeRepo(repo) | |||
| 73 | + | if err != nil { | |||
| 74 | + | return err | |||
| 75 | + | } | |||
| 76 | + | key := repoKey(repo.Name) | |||
| 77 | + | item, ok := s.db.Get(ctx, key) | |||
| 78 | + | if !ok { | |||
| 79 | + | return domain.ErrNotFound | |||
| 80 | + | } | |||
| 81 | + | var b kv.Batch | |||
| 82 | + | b.RequireVersion(key, item.Version) | |||
| 83 | + | b.Put(key, val) | |||
| 84 | + | if _, err := s.db.Apply(ctx, &b); err != nil { | |||
| 85 | + | if errors.Is(err, kv.ErrConflict) { | |||
| 86 | + | return domain.ErrConflict | |||
| 87 | + | } | |||
| 88 | + | return fmt.Errorf("update repository %q: %w", repo.Name, err) | |||
| 89 | + | } | |||
| 90 | + | return nil | |||
| 91 | + | } | |||
| 92 | + | ||||
| 93 | + | // Delete removes the repository; a missing one is not an error. | |||
| 94 | + | func (s *Store) Delete(ctx context.Context, name domain.RepoName) error { | |||
| 95 | + | var b kv.Batch | |||
| 96 | + | b.Delete(repoKey(name)) | |||
| 97 | + | if _, err := s.db.Apply(ctx, &b); err != nil { | |||
| 98 | + | return fmt.Errorf("delete repository %q: %w", name, err) | |||
| 99 | + | } | |||
| 100 | + | return nil | |||
| 101 | + | } | |||
| 102 | + | ||||
| 103 | + | func encodeRepo(r domain.Repo) ([]byte, error) { | |||
| 104 | + | if r.Name.IsZero() { | |||
| 105 | + | return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName) | |||
| 106 | + | } | |||
| 107 | + | if !utf8.ValidString(r.Description) { | |||
| 108 | + | return nil, errors.New("repository description is not valid UTF-8") | |||
| 109 | + | } | |||
| 110 | + | vis, err := encodeVisibility(r.Visibility) | |||
| 111 | + | if err != nil { | |||
| 112 | + | return nil, err | |||
| 113 | + | } | |||
| 114 | + | pb := &storagepb.Repo{ | |||
| 115 | + | Name: r.Name.String(), | |||
| 116 | + | Description: r.Description, | |||
| 117 | + | Visibility: vis, | |||
| 118 | + | CreatedAt: encodeTime(r.CreatedAt), | |||
| 119 | + | PushedAt: encodeTime(r.PushedAt), | |||
| 120 | + | } | |||
| 121 | + | return mustMarshal(pb), nil | |||
| 122 | + | } | |||
| 123 | + | ||||
| 124 | + | func decodeRepo(val []byte) (domain.Repo, error) { | |||
| 125 | + | var pb storagepb.Repo | |||
| 126 | + | if err := proto.Unmarshal(val, &pb); err != nil { | |||
| 127 | + | return domain.Repo{}, fmt.Errorf("decode repository: %w", err) | |||
| 128 | + | } | |||
| 129 | + | name, err := domain.ParseRepoName(pb.GetName()) | |||
| 130 | + | if err != nil { | |||
| 131 | + | return domain.Repo{}, fmt.Errorf("decode repository: %w", err) | |||
| 132 | + | } | |||
| 133 | + | return domain.Repo{ | |||
| 134 | + | Name: name, | |||
| 135 | + | Description: pb.GetDescription(), | |||
| 136 | + | Visibility: decodeVisibility(pb.GetVisibility()), | |||
| 137 | + | CreatedAt: decodeTime(pb.GetCreatedAt()), | |||
| 138 | + | PushedAt: decodeTime(pb.GetPushedAt()), | |||
| 139 | + | }, nil | |||
| 140 | + | } | |||
| 141 | + | ||||
| 142 | + | func encodeVisibility(v domain.Visibility) (storagepb.Visibility, error) { | |||
| 143 | + | switch v { | |||
| 144 | + | case domain.VisibilityPublic: | |||
| 145 | + | return storagepb.Visibility_VISIBILITY_PUBLIC, nil | |||
| 146 | + | case domain.VisibilityPrivate: | |||
| 147 | + | return storagepb.Visibility_VISIBILITY_PRIVATE, nil | |||
| 148 | + | } | |||
| 149 | + | return storagepb.Visibility_VISIBILITY_UNSPECIFIED, fmt.Errorf("%w: %q", domain.ErrInvalidVisibility, v) | |||
| 150 | + | } | |||
| 151 | + | ||||
| 152 | + | // decodeVisibility maps anything but public to private, including the | |||
| 153 | + | // unspecified value and values from a newer schema: the safe side. | |||
| 154 | + | func decodeVisibility(v storagepb.Visibility) domain.Visibility { | |||
| 155 | + | if v == storagepb.Visibility_VISIBILITY_PUBLIC { | |||
| 156 | + | return domain.VisibilityPublic | |||
| 157 | + | } | |||
| 158 | + | return domain.VisibilityPrivate | |||
| 159 | + | } | |||
| 160 | + | ||||
| 161 | + | func encodeTime(t time.Time) *timestamppb.Timestamp { | |||
| 162 | + | if t.IsZero() { | |||
| 163 | + | return nil | |||
| 164 | + | } | |||
| 165 | + | return timestamppb.New(t) | |||
| 166 | + | } | |||
| 167 | + | ||||
| 168 | + | func decodeTime(ts *timestamppb.Timestamp) time.Time { | |||
| 169 | + | if ts == nil { | |||
| 170 | + | return time.Time{} | |||
| 171 | + | } | |||
| 172 | + | return ts.AsTime() | |||
| 173 | + | } | |||
A internal/adapters/kvstore/store.go +110 -0
| @@ -0,0 +1,110 @@ | |||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |
| 2 | + | ||
| 3 | + | // Package kvstore implements the metadata ports over ermite-kv. Values are | |
| 4 | + | // the protobuf messages of proto/ermite/storage. | |
| 5 | + | package kvstore | |
| 6 | + | ||
| 7 | + | import ( | |
| 8 | + | "context" | |
| 9 | + | "errors" | |
| 10 | + | "fmt" | |
| 11 | + | ||
| 12 | + | kv "bnema.dev/ermite-kv" | |
| 13 | + | "github.com/bnema/zerowrap" | |
| 14 | + | "google.golang.org/protobuf/proto" | |
| 15 | + | ||
| 16 | + | storagepb "bnema.dev/ermite/gen/ermite/storage" | |
| 17 | + | "bnema.dev/ermite/internal/ports" | |
| 18 | + | ) | |
| 19 | + | ||
| 20 | + | // SchemaVersion is the storage schema this build reads and writes. | |
| 21 | + | const SchemaVersion = 1 | |
| 22 | + | ||
| 23 | + | const schemaKey = "meta/schema" | |
| 24 | + | ||
| 25 | + | // ErrSchemaTooNew means the data was written by a newer ermite. | |
| 26 | + | var ErrSchemaTooNew = errors.New("storage schema is newer than this ermite supports") | |
| 27 | + | ||
| 28 | + | // Store is the ermite-kv backed metadata store. | |
| 29 | + | type Store struct { | |
| 30 | + | db *kv.DB | |
| 31 | + | } | |
| 32 | + | ||
| 33 | + | var _ ports.RepoMetaStore = (*Store)(nil) | |
| 34 | + | ||
| 35 | + | // Open opens (or creates) the database in dir and checks the storage schema: | |
| 36 | + | // a missing version is written as SchemaVersion, a higher one stops with | |
| 37 | + | // ErrSchemaTooNew. dir must exist. | |
| 38 | + | func Open(ctx context.Context, dir string) (*Store, error) { | |
| 39 | + | db, err := kv.Open(ctx, dir, kv.Options{}) | |
| 40 | + | if err != nil { | |
| 41 | + | return nil, fmt.Errorf("open ermite-kv in %s: %w", dir, err) | |
| 42 | + | } | |
| 43 | + | s := &Store{db: db} | |
| 44 | + | if err := s.checkSchema(ctx); err != nil { | |
| 45 | + | _ = db.Close(context.WithoutCancel(ctx)) | |
| 46 | + | return nil, err | |
| 47 | + | } | |
| 48 | + | return s, nil | |
| 49 | + | } | |
| 50 | + | ||
| 51 | + | // Close stops the database after the writes in progress. | |
| 52 | + | func (s *Store) Close(ctx context.Context) error { | |
| 53 | + | if err := s.db.Close(ctx); err != nil { | |
| 54 | + | return fmt.Errorf("close ermite-kv: %w", err) | |
| 55 | + | } | |
| 56 | + | return nil | |
| 57 | + | } | |
| 58 | + | ||
| 59 | + | // Check reports whether the database accepts writes. It returns the failure | |
| 60 | + | // once ermite-kv is closed or failed, and writes nothing. | |
| 61 | + | func (s *Store) Check(ctx context.Context) error { | |
| 62 | + | var b kv.Batch // no ops: only the state of the database is checked | |
| 63 | + | if _, err := s.db.Apply(ctx, &b); err != nil { | |
| 64 | + | return fmt.Errorf("ermite-kv: %w", err) | |
| 65 | + | } | |
| 66 | + | return nil | |
| 67 | + | } | |
| 68 | + | ||
| 69 | + | func (s *Store) checkSchema(ctx context.Context) error { | |
| 70 | + | log := zerowrap.FromCtx(ctx) | |
| 71 | + | key := []byte(schemaKey) | |
| 72 | + | for range 2 { | |
| 73 | + | item, ok := s.db.Get(ctx, key) | |
| 74 | + | if !ok { | |
| 75 | + | var b kv.Batch | |
| 76 | + | b.RequireAbsent(key) | |
| 77 | + | b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: SchemaVersion})) | |
| 78 | + | _, err := s.db.Apply(ctx, &b) | |
| 79 | + | if errors.Is(err, kv.ErrConflict) { | |
| 80 | + | continue // written meanwhile: read it | |
| 81 | + | } | |
| 82 | + | if err != nil { | |
| 83 | + | return fmt.Errorf("write storage schema: %w", err) | |
| 84 | + | } | |
| 85 | + | log.Info().Uint32("schema", SchemaVersion).Msg("storage schema initialized") | |
| 86 | + | return nil | |
| 87 | + | } | |
| 88 | + | var sv storagepb.SchemaVersion | |
| 89 | + | if err := proto.Unmarshal(item.Value, &sv); err != nil { | |
| 90 | + | return fmt.Errorf("decode storage schema: %w", err) | |
| 91 | + | } | |
| 92 | + | switch { | |
| 93 | + | case sv.GetVersion() > SchemaVersion: | |
| 94 | + | return fmt.Errorf("%w: found %d, supported %d", ErrSchemaTooNew, sv.GetVersion(), SchemaVersion) | |
| 95 | + | case sv.GetVersion() == 0: | |
| 96 | + | return fmt.Errorf("decode storage schema: invalid version 0") | |
| 97 | + | } | |
| 98 | + | log.Info().Uint32("schema", sv.GetVersion()).Msg("storage schema checked") | |
| 99 | + | return nil | |
| 100 | + | } | |
| 101 | + | return errors.New("write storage schema: concurrent initialization did not settle") | |
| 102 | + | } | |
| 103 | + | ||
| 104 | + | func mustMarshal(m proto.Message) []byte { | |
| 105 | + | b, err := proto.MarshalOptions{Deterministic: true}.Marshal(m) | |
| 106 | + | if err != nil { | |
| 107 | + | panic(fmt.Sprintf("kvstore: marshal %T: %v", m, err)) // only for invalid UTF-8 strings, checked by callers | |
| 108 | + | } | |
| 109 | + | return b | |
| 110 | + | } | |
| @@ -0,0 +1,110 @@ | |||||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |||
| 2 | + | ||||
| 3 | + | // Package kvstore implements the metadata ports over ermite-kv. Values are | |||
| 4 | + | // the protobuf messages of proto/ermite/storage. | |||
| 5 | + | package kvstore | |||
| 6 | + | ||||
| 7 | + | import ( | |||
| 8 | + | "context" | |||
| 9 | + | "errors" | |||
| 10 | + | "fmt" | |||
| 11 | + | ||||
| 12 | + | kv "bnema.dev/ermite-kv" | |||
| 13 | + | "github.com/bnema/zerowrap" | |||
| 14 | + | "google.golang.org/protobuf/proto" | |||
| 15 | + | ||||
| 16 | + | storagepb "bnema.dev/ermite/gen/ermite/storage" | |||
| 17 | + | "bnema.dev/ermite/internal/ports" | |||
| 18 | + | ) | |||
| 19 | + | ||||
| 20 | + | // SchemaVersion is the storage schema this build reads and writes. | |||
| 21 | + | const SchemaVersion = 1 | |||
| 22 | + | ||||
| 23 | + | const schemaKey = "meta/schema" | |||
| 24 | + | ||||
| 25 | + | // ErrSchemaTooNew means the data was written by a newer ermite. | |||
| 26 | + | var ErrSchemaTooNew = errors.New("storage schema is newer than this ermite supports") | |||
| 27 | + | ||||
| 28 | + | // Store is the ermite-kv backed metadata store. | |||
| 29 | + | type Store struct { | |||
| 30 | + | db *kv.DB | |||
| 31 | + | } | |||
| 32 | + | ||||
| 33 | + | var _ ports.RepoMetaStore = (*Store)(nil) | |||
| 34 | + | ||||
| 35 | + | // Open opens (or creates) the database in dir and checks the storage schema: | |||
| 36 | + | // a missing version is written as SchemaVersion, a higher one stops with | |||
| 37 | + | // ErrSchemaTooNew. dir must exist. | |||
| 38 | + | func Open(ctx context.Context, dir string) (*Store, error) { | |||
| 39 | + | db, err := kv.Open(ctx, dir, kv.Options{}) | |||
| 40 | + | if err != nil { | |||
| 41 | + | return nil, fmt.Errorf("open ermite-kv in %s: %w", dir, err) | |||
| 42 | + | } | |||
| 43 | + | s := &Store{db: db} | |||
| 44 | + | if err := s.checkSchema(ctx); err != nil { | |||
| 45 | + | _ = db.Close(context.WithoutCancel(ctx)) | |||
| 46 | + | return nil, err | |||
| 47 | + | } | |||
| 48 | + | return s, nil | |||
| 49 | + | } | |||
| 50 | + | ||||
| 51 | + | // Close stops the database after the writes in progress. | |||
| 52 | + | func (s *Store) Close(ctx context.Context) error { | |||
| 53 | + | if err := s.db.Close(ctx); err != nil { | |||
| 54 | + | return fmt.Errorf("close ermite-kv: %w", err) | |||
| 55 | + | } | |||
| 56 | + | return nil | |||
| 57 | + | } | |||
| 58 | + | ||||
| 59 | + | // Check reports whether the database accepts writes. It returns the failure | |||
| 60 | + | // once ermite-kv is closed or failed, and writes nothing. | |||
| 61 | + | func (s *Store) Check(ctx context.Context) error { | |||
| 62 | + | var b kv.Batch // no ops: only the state of the database is checked | |||
| 63 | + | if _, err := s.db.Apply(ctx, &b); err != nil { | |||
| 64 | + | return fmt.Errorf("ermite-kv: %w", err) | |||
| 65 | + | } | |||
| 66 | + | return nil | |||
| 67 | + | } | |||
| 68 | + | ||||
| 69 | + | func (s *Store) checkSchema(ctx context.Context) error { | |||
| 70 | + | log := zerowrap.FromCtx(ctx) | |||
| 71 | + | key := []byte(schemaKey) | |||
| 72 | + | for range 2 { | |||
| 73 | + | item, ok := s.db.Get(ctx, key) | |||
| 74 | + | if !ok { | |||
| 75 | + | var b kv.Batch | |||
| 76 | + | b.RequireAbsent(key) | |||
| 77 | + | b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: SchemaVersion})) | |||
| 78 | + | _, err := s.db.Apply(ctx, &b) | |||
| 79 | + | if errors.Is(err, kv.ErrConflict) { | |||
| 80 | + | continue // written meanwhile: read it | |||
| 81 | + | } | |||
| 82 | + | if err != nil { | |||
| 83 | + | return fmt.Errorf("write storage schema: %w", err) | |||
| 84 | + | } | |||
| 85 | + | log.Info().Uint32("schema", SchemaVersion).Msg("storage schema initialized") | |||
| 86 | + | return nil | |||
| 87 | + | } | |||
| 88 | + | var sv storagepb.SchemaVersion | |||
| 89 | + | if err := proto.Unmarshal(item.Value, &sv); err != nil { | |||
| 90 | + | return fmt.Errorf("decode storage schema: %w", err) | |||
| 91 | + | } | |||
| 92 | + | switch { | |||
| 93 | + | case sv.GetVersion() > SchemaVersion: | |||
| 94 | + | return fmt.Errorf("%w: found %d, supported %d", ErrSchemaTooNew, sv.GetVersion(), SchemaVersion) | |||
| 95 | + | case sv.GetVersion() == 0: | |||
| 96 | + | return fmt.Errorf("decode storage schema: invalid version 0") | |||
| 97 | + | } | |||
| 98 | + | log.Info().Uint32("schema", sv.GetVersion()).Msg("storage schema checked") | |||
| 99 | + | return nil | |||
| 100 | + | } | |||
| 101 | + | return errors.New("write storage schema: concurrent initialization did not settle") | |||
| 102 | + | } | |||
| 103 | + | ||||
| 104 | + | func mustMarshal(m proto.Message) []byte { | |||
| 105 | + | b, err := proto.MarshalOptions{Deterministic: true}.Marshal(m) | |||
| 106 | + | if err != nil { | |||
| 107 | + | panic(fmt.Sprintf("kvstore: marshal %T: %v", m, err)) // only for invalid UTF-8 strings, checked by callers | |||
| 108 | + | } | |||
| 109 | + | return b | |||
| 110 | + | } | |||
A internal/adapters/kvstore/store_test.go +262 -0
| @@ -0,0 +1,262 @@ | |||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |
| 2 | + | ||
| 3 | + | package kvstore_test | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "context" | |
| 7 | + | "errors" | |
| 8 | + | "fmt" | |
| 9 | + | "sync" | |
| 10 | + | "testing" | |
| 11 | + | "time" | |
| 12 | + | ||
| 13 | + | kv "bnema.dev/ermite-kv" | |
| 14 | + | "github.com/stretchr/testify/assert" | |
| 15 | + | "github.com/stretchr/testify/require" | |
| 16 | + | "google.golang.org/protobuf/proto" | |
| 17 | + | ||
| 18 | + | storagepb "bnema.dev/ermite/gen/ermite/storage" | |
| 19 | + | "bnema.dev/ermite/internal/adapters/kvstore" | |
| 20 | + | "bnema.dev/ermite/internal/domain" | |
| 21 | + | ) | |
| 22 | + | ||
| 23 | + | func openStore(t *testing.T, dir string) *kvstore.Store { | |
| 24 | + | t.Helper() | |
| 25 | + | ctx := context.Background() | |
| 26 | + | s, err := kvstore.Open(ctx, dir) | |
| 27 | + | require.NoError(t, err) | |
| 28 | + | t.Cleanup(func() { _ = s.Close(ctx) }) | |
| 29 | + | return s | |
| 30 | + | } | |
| 31 | + | ||
| 32 | + | func name(t *testing.T, s string) domain.RepoName { | |
| 33 | + | t.Helper() | |
| 34 | + | n, err := domain.ParseRepoName(s) | |
| 35 | + | require.NoError(t, err) | |
| 36 | + | return n | |
| 37 | + | } | |
| 38 | + | ||
| 39 | + | var created = time.Date(2026, 10, 10, 8, 0, 0, 0, time.UTC) | |
| 40 | + | ||
| 41 | + | func TestCreateGetListUpdateDelete(t *testing.T) { | |
| 42 | + | ctx := context.Background() | |
| 43 | + | s := openStore(t, t.TempDir()) | |
| 44 | + | ||
| 45 | + | b := domain.Repo{Name: name(t, "b"), Description: "second", Visibility: domain.VisibilityPublic, CreatedAt: created} | |
| 46 | + | a := domain.Repo{Name: name(t, "a"), Visibility: domain.VisibilityPrivate, CreatedAt: created, PushedAt: created.Add(time.Hour)} | |
| 47 | + | require.NoError(t, s.Create(ctx, b)) | |
| 48 | + | require.NoError(t, s.Create(ctx, a)) | |
| 49 | + | ||
| 50 | + | got, err := s.Get(ctx, b.Name) | |
| 51 | + | require.NoError(t, err) | |
| 52 | + | assert.Equal(t, b, got) | |
| 53 | + | got, err = s.Get(ctx, a.Name) | |
| 54 | + | require.NoError(t, err) | |
| 55 | + | assert.Equal(t, a, got) | |
| 56 | + | assert.True(t, got.PushedAt.Equal(a.PushedAt)) | |
| 57 | + | ||
| 58 | + | list, err := s.List(ctx) | |
| 59 | + | require.NoError(t, err) | |
| 60 | + | assert.Equal(t, []domain.Repo{a, b}, list) | |
| 61 | + | ||
| 62 | + | err = s.Create(ctx, b) | |
| 63 | + | assert.ErrorIs(t, err, domain.ErrAlreadyExists) | |
| 64 | + | ||
| 65 | + | b.Description = "changed" | |
| 66 | + | b.Visibility = domain.VisibilityPrivate | |
| 67 | + | require.NoError(t, s.Update(ctx, b)) | |
| 68 | + | got, err = s.Get(ctx, b.Name) | |
| 69 | + | require.NoError(t, err) | |
| 70 | + | assert.Equal(t, b, got) | |
| 71 | + | ||
| 72 | + | require.NoError(t, s.Delete(ctx, b.Name)) | |
| 73 | + | _, err = s.Get(ctx, b.Name) | |
| 74 | + | assert.ErrorIs(t, err, domain.ErrNotFound) | |
| 75 | + | require.NoError(t, s.Delete(ctx, b.Name), "deleting a missing repository is not an error") | |
| 76 | + | assert.ErrorIs(t, s.Update(ctx, b), domain.ErrNotFound) | |
| 77 | + | ||
| 78 | + | _, err = s.Get(ctx, name(t, "never")) | |
| 79 | + | assert.ErrorIs(t, err, domain.ErrNotFound) | |
| 80 | + | } | |
| 81 | + | ||
| 82 | + | func TestCreateRejectsInvalidValues(t *testing.T) { | |
| 83 | + | ctx := context.Background() | |
| 84 | + | s := openStore(t, t.TempDir()) | |
| 85 | + | n := name(t, "x") | |
| 86 | + | ||
| 87 | + | assert.ErrorIs(t, s.Create(ctx, domain.Repo{Name: n, Visibility: "secret"}), domain.ErrInvalidVisibility) | |
| 88 | + | assert.ErrorIs(t, s.Create(ctx, domain.Repo{Visibility: domain.VisibilityPublic}), domain.ErrInvalidName) | |
| 89 | + | assert.Error(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPublic, Description: "\xff"})) | |
| 90 | + | list, err := s.List(ctx) | |
| 91 | + | require.NoError(t, err) | |
| 92 | + | assert.Empty(t, list) | |
| 93 | + | } | |
| 94 | + | ||
| 95 | + | func TestConcurrentCreateSameName(t *testing.T) { | |
| 96 | + | ctx := context.Background() | |
| 97 | + | s := openStore(t, t.TempDir()) | |
| 98 | + | n := name(t, "tools") | |
| 99 | + | ||
| 100 | + | const workers = 16 | |
| 101 | + | errs := make(chan error, workers) | |
| 102 | + | var wg sync.WaitGroup | |
| 103 | + | for i := range workers { | |
| 104 | + | wg.Add(1) | |
| 105 | + | go func() { | |
| 106 | + | defer wg.Done() | |
| 107 | + | errs <- s.Create(ctx, domain.Repo{Name: n, Description: fmt.Sprint(i), Visibility: domain.VisibilityPrivate, CreatedAt: created}) | |
| 108 | + | }() | |
| 109 | + | } | |
| 110 | + | wg.Wait() | |
| 111 | + | close(errs) | |
| 112 | + | ok, exists := 0, 0 | |
| 113 | + | for err := range errs { | |
| 114 | + | switch { | |
| 115 | + | case err == nil: | |
| 116 | + | ok++ | |
| 117 | + | case errors.Is(err, domain.ErrAlreadyExists): | |
| 118 | + | exists++ | |
| 119 | + | default: | |
| 120 | + | t.Fatalf("unexpected error: %v", err) | |
| 121 | + | } | |
| 122 | + | } | |
| 123 | + | assert.Equal(t, 1, ok) | |
| 124 | + | assert.Equal(t, workers-1, exists) | |
| 125 | + | } | |
| 126 | + | ||
| 127 | + | func TestPersistsAcrossReopen(t *testing.T) { | |
| 128 | + | ctx := context.Background() | |
| 129 | + | dir := t.TempDir() | |
| 130 | + | s, err := kvstore.Open(ctx, dir) | |
| 131 | + | require.NoError(t, err) | |
| 132 | + | r := domain.Repo{Name: name(t, "keep"), Visibility: domain.VisibilityPublic, CreatedAt: created} | |
| 133 | + | require.NoError(t, s.Create(ctx, r)) | |
| 134 | + | require.NoError(t, s.Close(ctx)) | |
| 135 | + | ||
| 136 | + | s2 := openStore(t, dir) | |
| 137 | + | got, err := s2.Get(ctx, r.Name) | |
| 138 | + | require.NoError(t, err) | |
| 139 | + | assert.Equal(t, r, got) | |
| 140 | + | } | |
| 141 | + | ||
| 142 | + | func TestCheck(t *testing.T) { | |
| 143 | + | ctx := context.Background() | |
| 144 | + | s, err := kvstore.Open(ctx, t.TempDir()) | |
| 145 | + | require.NoError(t, err) | |
| 146 | + | require.NoError(t, s.Check(ctx)) | |
| 147 | + | require.NoError(t, s.Close(ctx)) | |
| 148 | + | assert.ErrorIs(t, s.Check(ctx), kv.ErrClosed) | |
| 149 | + | } | |
| 150 | + | ||
| 151 | + | func TestOpenLockedDirectory(t *testing.T) { | |
| 152 | + | ctx := context.Background() | |
| 153 | + | dir := t.TempDir() | |
| 154 | + | _ = openStore(t, dir) | |
| 155 | + | _, err := kvstore.Open(ctx, dir) | |
| 156 | + | assert.ErrorIs(t, err, kv.ErrLocked) | |
| 157 | + | } | |
| 158 | + | ||
| 159 | + | // putRaw writes a value with ermite-kv directly, bypassing the store. | |
| 160 | + | func putRaw(t *testing.T, dir, key string, val []byte) { | |
| 161 | + | t.Helper() | |
| 162 | + | ctx := context.Background() | |
| 163 | + | db, err := kv.Open(ctx, dir, kv.Options{}) | |
| 164 | + | require.NoError(t, err) | |
| 165 | + | var b kv.Batch | |
| 166 | + | b.Put([]byte(key), val) | |
| 167 | + | _, err = db.Apply(ctx, &b) | |
| 168 | + | require.NoError(t, err) | |
| 169 | + | require.NoError(t, db.Close(ctx)) | |
| 170 | + | } | |
| 171 | + | ||
| 172 | + | func readSchema(t *testing.T, dir string) uint32 { | |
| 173 | + | t.Helper() | |
| 174 | + | ctx := context.Background() | |
| 175 | + | db, err := kv.Open(ctx, dir, kv.Options{}) | |
| 176 | + | require.NoError(t, err) | |
| 177 | + | defer func() { _ = db.Close(ctx) }() | |
| 178 | + | item, ok := db.Get(ctx, []byte("meta/schema")) | |
| 179 | + | require.True(t, ok, "meta/schema missing") | |
| 180 | + | var sv storagepb.SchemaVersion | |
| 181 | + | require.NoError(t, proto.Unmarshal(item.Value, &sv)) | |
| 182 | + | return sv.GetVersion() | |
| 183 | + | } | |
| 184 | + | ||
| 185 | + | func TestSchema(t *testing.T) { | |
| 186 | + | ctx := context.Background() | |
| 187 | + | ||
| 188 | + | t.Run("missing is written as 1", func(t *testing.T) { | |
| 189 | + | dir := t.TempDir() | |
| 190 | + | s, err := kvstore.Open(ctx, dir) | |
| 191 | + | require.NoError(t, err) | |
| 192 | + | require.NoError(t, s.Close(ctx)) | |
| 193 | + | assert.Equal(t, uint32(1), readSchema(t, dir)) | |
| 194 | + | }) | |
| 195 | + | ||
| 196 | + | t.Run("current is accepted and kept", func(t *testing.T) { | |
| 197 | + | dir := t.TempDir() | |
| 198 | + | s, err := kvstore.Open(ctx, dir) | |
| 199 | + | require.NoError(t, err) | |
| 200 | + | require.NoError(t, s.Close(ctx)) | |
| 201 | + | s, err = kvstore.Open(ctx, dir) | |
| 202 | + | require.NoError(t, err) | |
| 203 | + | require.NoError(t, s.Close(ctx)) | |
| 204 | + | assert.Equal(t, uint32(1), readSchema(t, dir)) | |
| 205 | + | }) | |
| 206 | + | ||
| 207 | + | t.Run("newer stops startup and releases the lock", func(t *testing.T) { | |
| 208 | + | dir := t.TempDir() | |
| 209 | + | putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 2})) | |
| 210 | + | _, err := kvstore.Open(ctx, dir) | |
| 211 | + | require.ErrorIs(t, err, kvstore.ErrSchemaTooNew) | |
| 212 | + | assert.Equal(t, uint32(2), readSchema(t, dir), "the lock was released and the value kept") | |
| 213 | + | }) | |
| 214 | + | ||
| 215 | + | t.Run("garbage and zero are errors", func(t *testing.T) { | |
| 216 | + | dir := t.TempDir() | |
| 217 | + | putRaw(t, dir, "meta/schema", []byte{0xff, 0xff, 0xff}) | |
| 218 | + | _, err := kvstore.Open(ctx, dir) | |
| 219 | + | assert.Error(t, err) | |
| 220 | + | ||
| 221 | + | dir = t.TempDir() | |
| 222 | + | putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 0})) | |
| 223 | + | _, err = kvstore.Open(ctx, dir) | |
| 224 | + | assert.Error(t, err) | |
| 225 | + | }) | |
| 226 | + | } | |
| 227 | + | ||
| 228 | + | func mustProto(t *testing.T, m proto.Message) []byte { | |
| 229 | + | t.Helper() | |
| 230 | + | b, err := proto.Marshal(m) | |
| 231 | + | require.NoError(t, err) | |
| 232 | + | return b | |
| 233 | + | } | |
| 234 | + | ||
| 235 | + | func TestUnspecifiedVisibilityReadsPrivate(t *testing.T) { | |
| 236 | + | ctx := context.Background() | |
| 237 | + | dir := t.TempDir() | |
| 238 | + | putRaw(t, dir, "repo/odd", mustProto(t, &storagepb.Repo{Name: "odd"})) // visibility unset | |
| 239 | + | putRaw(t, dir, "repo/future", mustProto(t, &storagepb.Repo{Name: "future", Visibility: storagepb.Visibility(42)})) | |
| 240 | + | s := openStore(t, dir) | |
| 241 | + | ||
| 242 | + | for _, n := range []string{"odd", "future"} { | |
| 243 | + | got, err := s.Get(ctx, name(t, n)) | |
| 244 | + | require.NoError(t, err) | |
| 245 | + | assert.Equal(t, domain.VisibilityPrivate, got.Visibility, n) | |
| 246 | + | } | |
| 247 | + | } | |
| 248 | + | ||
| 249 | + | func TestCorruptRepoValue(t *testing.T) { | |
| 250 | + | ctx := context.Background() | |
| 251 | + | dir := t.TempDir() | |
| 252 | + | putRaw(t, dir, "repo/bad", []byte{0xff, 0xff}) | |
| 253 | + | putRaw(t, dir, "repo/mismatch", mustProto(t, &storagepb.Repo{Name: "Not Valid"})) | |
| 254 | + | s := openStore(t, dir) | |
| 255 | + | ||
| 256 | + | _, err := s.Get(ctx, name(t, "bad")) | |
| 257 | + | assert.Error(t, err) | |
| 258 | + | _, err = s.Get(ctx, name(t, "mismatch")) | |
| 259 | + | assert.ErrorIs(t, err, domain.ErrInvalidName) | |
| 260 | + | _, err = s.List(ctx) | |
| 261 | + | assert.Error(t, err) | |
| 262 | + | } | |
| @@ -0,0 +1,262 @@ | |||||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |||
| 2 | + | ||||
| 3 | + | package kvstore_test | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "context" | |||
| 7 | + | "errors" | |||
| 8 | + | "fmt" | |||
| 9 | + | "sync" | |||
| 10 | + | "testing" | |||
| 11 | + | "time" | |||
| 12 | + | ||||
| 13 | + | kv "bnema.dev/ermite-kv" | |||
| 14 | + | "github.com/stretchr/testify/assert" | |||
| 15 | + | "github.com/stretchr/testify/require" | |||
| 16 | + | "google.golang.org/protobuf/proto" | |||
| 17 | + | ||||
| 18 | + | storagepb "bnema.dev/ermite/gen/ermite/storage" | |||
| 19 | + | "bnema.dev/ermite/internal/adapters/kvstore" | |||
| 20 | + | "bnema.dev/ermite/internal/domain" | |||
| 21 | + | ) | |||
| 22 | + | ||||
| 23 | + | func openStore(t *testing.T, dir string) *kvstore.Store { | |||
| 24 | + | t.Helper() | |||
| 25 | + | ctx := context.Background() | |||
| 26 | + | s, err := kvstore.Open(ctx, dir) | |||
| 27 | + | require.NoError(t, err) | |||
| 28 | + | t.Cleanup(func() { _ = s.Close(ctx) }) | |||
| 29 | + | return s | |||
| 30 | + | } | |||
| 31 | + | ||||
| 32 | + | func name(t *testing.T, s string) domain.RepoName { | |||
| 33 | + | t.Helper() | |||
| 34 | + | n, err := domain.ParseRepoName(s) | |||
| 35 | + | require.NoError(t, err) | |||
| 36 | + | return n | |||
| 37 | + | } | |||
| 38 | + | ||||
| 39 | + | var created = time.Date(2026, 10, 10, 8, 0, 0, 0, time.UTC) | |||
| 40 | + | ||||
| 41 | + | func TestCreateGetListUpdateDelete(t *testing.T) { | |||
| 42 | + | ctx := context.Background() | |||
| 43 | + | s := openStore(t, t.TempDir()) | |||
| 44 | + | ||||
| 45 | + | b := domain.Repo{Name: name(t, "b"), Description: "second", Visibility: domain.VisibilityPublic, CreatedAt: created} | |||
| 46 | + | a := domain.Repo{Name: name(t, "a"), Visibility: domain.VisibilityPrivate, CreatedAt: created, PushedAt: created.Add(time.Hour)} | |||
| 47 | + | require.NoError(t, s.Create(ctx, b)) | |||
| 48 | + | require.NoError(t, s.Create(ctx, a)) | |||
| 49 | + | ||||
| 50 | + | got, err := s.Get(ctx, b.Name) | |||
| 51 | + | require.NoError(t, err) | |||
| 52 | + | assert.Equal(t, b, got) | |||
| 53 | + | got, err = s.Get(ctx, a.Name) | |||
| 54 | + | require.NoError(t, err) | |||
| 55 | + | assert.Equal(t, a, got) | |||
| 56 | + | assert.True(t, got.PushedAt.Equal(a.PushedAt)) | |||
| 57 | + | ||||
| 58 | + | list, err := s.List(ctx) | |||
| 59 | + | require.NoError(t, err) | |||
| 60 | + | assert.Equal(t, []domain.Repo{a, b}, list) | |||
| 61 | + | ||||
| 62 | + | err = s.Create(ctx, b) | |||
| 63 | + | assert.ErrorIs(t, err, domain.ErrAlreadyExists) | |||
| 64 | + | ||||
| 65 | + | b.Description = "changed" | |||
| 66 | + | b.Visibility = domain.VisibilityPrivate | |||
| 67 | + | require.NoError(t, s.Update(ctx, b)) | |||
| 68 | + | got, err = s.Get(ctx, b.Name) | |||
| 69 | + | require.NoError(t, err) | |||
| 70 | + | assert.Equal(t, b, got) | |||
| 71 | + | ||||
| 72 | + | require.NoError(t, s.Delete(ctx, b.Name)) | |||
| 73 | + | _, err = s.Get(ctx, b.Name) | |||
| 74 | + | assert.ErrorIs(t, err, domain.ErrNotFound) | |||
| 75 | + | require.NoError(t, s.Delete(ctx, b.Name), "deleting a missing repository is not an error") | |||
| 76 | + | assert.ErrorIs(t, s.Update(ctx, b), domain.ErrNotFound) | |||
| 77 | + | ||||
| 78 | + | _, err = s.Get(ctx, name(t, "never")) | |||
| 79 | + | assert.ErrorIs(t, err, domain.ErrNotFound) | |||
| 80 | + | } | |||
| 81 | + | ||||
| 82 | + | func TestCreateRejectsInvalidValues(t *testing.T) { | |||
| 83 | + | ctx := context.Background() | |||
| 84 | + | s := openStore(t, t.TempDir()) | |||
| 85 | + | n := name(t, "x") | |||
| 86 | + | ||||
| 87 | + | assert.ErrorIs(t, s.Create(ctx, domain.Repo{Name: n, Visibility: "secret"}), domain.ErrInvalidVisibility) | |||
| 88 | + | assert.ErrorIs(t, s.Create(ctx, domain.Repo{Visibility: domain.VisibilityPublic}), domain.ErrInvalidName) | |||
| 89 | + | assert.Error(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPublic, Description: "\xff"})) | |||
| 90 | + | list, err := s.List(ctx) | |||
| 91 | + | require.NoError(t, err) | |||
| 92 | + | assert.Empty(t, list) | |||
| 93 | + | } | |||
| 94 | + | ||||
| 95 | + | func TestConcurrentCreateSameName(t *testing.T) { | |||
| 96 | + | ctx := context.Background() | |||
| 97 | + | s := openStore(t, t.TempDir()) | |||
| 98 | + | n := name(t, "tools") | |||
| 99 | + | ||||
| 100 | + | const workers = 16 | |||
| 101 | + | errs := make(chan error, workers) | |||
| 102 | + | var wg sync.WaitGroup | |||
| 103 | + | for i := range workers { | |||
| 104 | + | wg.Add(1) | |||
| 105 | + | go func() { | |||
| 106 | + | defer wg.Done() | |||
| 107 | + | errs <- s.Create(ctx, domain.Repo{Name: n, Description: fmt.Sprint(i), Visibility: domain.VisibilityPrivate, CreatedAt: created}) | |||
| 108 | + | }() | |||
| 109 | + | } | |||
| 110 | + | wg.Wait() | |||
| 111 | + | close(errs) | |||
| 112 | + | ok, exists := 0, 0 | |||
| 113 | + | for err := range errs { | |||
| 114 | + | switch { | |||
| 115 | + | case err == nil: | |||
| 116 | + | ok++ | |||
| 117 | + | case errors.Is(err, domain.ErrAlreadyExists): | |||
| 118 | + | exists++ | |||
| 119 | + | default: | |||
| 120 | + | t.Fatalf("unexpected error: %v", err) | |||
| 121 | + | } | |||
| 122 | + | } | |||
| 123 | + | assert.Equal(t, 1, ok) | |||
| 124 | + | assert.Equal(t, workers-1, exists) | |||
| 125 | + | } | |||
| 126 | + | ||||
| 127 | + | func TestPersistsAcrossReopen(t *testing.T) { | |||
| 128 | + | ctx := context.Background() | |||
| 129 | + | dir := t.TempDir() | |||
| 130 | + | s, err := kvstore.Open(ctx, dir) | |||
| 131 | + | require.NoError(t, err) | |||
| 132 | + | r := domain.Repo{Name: name(t, "keep"), Visibility: domain.VisibilityPublic, CreatedAt: created} | |||
| 133 | + | require.NoError(t, s.Create(ctx, r)) | |||
| 134 | + | require.NoError(t, s.Close(ctx)) | |||
| 135 | + | ||||
| 136 | + | s2 := openStore(t, dir) | |||
| 137 | + | got, err := s2.Get(ctx, r.Name) | |||
| 138 | + | require.NoError(t, err) | |||
| 139 | + | assert.Equal(t, r, got) | |||
| 140 | + | } | |||
| 141 | + | ||||
| 142 | + | func TestCheck(t *testing.T) { | |||
| 143 | + | ctx := context.Background() | |||
| 144 | + | s, err := kvstore.Open(ctx, t.TempDir()) | |||
| 145 | + | require.NoError(t, err) | |||
| 146 | + | require.NoError(t, s.Check(ctx)) | |||
| 147 | + | require.NoError(t, s.Close(ctx)) | |||
| 148 | + | assert.ErrorIs(t, s.Check(ctx), kv.ErrClosed) | |||
| 149 | + | } | |||
| 150 | + | ||||
| 151 | + | func TestOpenLockedDirectory(t *testing.T) { | |||
| 152 | + | ctx := context.Background() | |||
| 153 | + | dir := t.TempDir() | |||
| 154 | + | _ = openStore(t, dir) | |||
| 155 | + | _, err := kvstore.Open(ctx, dir) | |||
| 156 | + | assert.ErrorIs(t, err, kv.ErrLocked) | |||
| 157 | + | } | |||
| 158 | + | ||||
| 159 | + | // putRaw writes a value with ermite-kv directly, bypassing the store. | |||
| 160 | + | func putRaw(t *testing.T, dir, key string, val []byte) { | |||
| 161 | + | t.Helper() | |||
| 162 | + | ctx := context.Background() | |||
| 163 | + | db, err := kv.Open(ctx, dir, kv.Options{}) | |||
| 164 | + | require.NoError(t, err) | |||
| 165 | + | var b kv.Batch | |||
| 166 | + | b.Put([]byte(key), val) | |||
| 167 | + | _, err = db.Apply(ctx, &b) | |||
| 168 | + | require.NoError(t, err) | |||
| 169 | + | require.NoError(t, db.Close(ctx)) | |||
| 170 | + | } | |||
| 171 | + | ||||
| 172 | + | func readSchema(t *testing.T, dir string) uint32 { | |||
| 173 | + | t.Helper() | |||
| 174 | + | ctx := context.Background() | |||
| 175 | + | db, err := kv.Open(ctx, dir, kv.Options{}) | |||
| 176 | + | require.NoError(t, err) | |||
| 177 | + | defer func() { _ = db.Close(ctx) }() | |||
| 178 | + | item, ok := db.Get(ctx, []byte("meta/schema")) | |||
| 179 | + | require.True(t, ok, "meta/schema missing") | |||
| 180 | + | var sv storagepb.SchemaVersion | |||
| 181 | + | require.NoError(t, proto.Unmarshal(item.Value, &sv)) | |||
| 182 | + | return sv.GetVersion() | |||
| 183 | + | } | |||
| 184 | + | ||||
| 185 | + | func TestSchema(t *testing.T) { | |||
| 186 | + | ctx := context.Background() | |||
| 187 | + | ||||
| 188 | + | t.Run("missing is written as 1", func(t *testing.T) { | |||
| 189 | + | dir := t.TempDir() | |||
| 190 | + | s, err := kvstore.Open(ctx, dir) | |||
| 191 | + | require.NoError(t, err) | |||
| 192 | + | require.NoError(t, s.Close(ctx)) | |||
| 193 | + | assert.Equal(t, uint32(1), readSchema(t, dir)) | |||
| 194 | + | }) | |||
| 195 | + | ||||
| 196 | + | t.Run("current is accepted and kept", func(t *testing.T) { | |||
| 197 | + | dir := t.TempDir() | |||
| 198 | + | s, err := kvstore.Open(ctx, dir) | |||
| 199 | + | require.NoError(t, err) | |||
| 200 | + | require.NoError(t, s.Close(ctx)) | |||
| 201 | + | s, err = kvstore.Open(ctx, dir) | |||
| 202 | + | require.NoError(t, err) | |||
| 203 | + | require.NoError(t, s.Close(ctx)) | |||
| 204 | + | assert.Equal(t, uint32(1), readSchema(t, dir)) | |||
| 205 | + | }) | |||
| 206 | + | ||||
| 207 | + | t.Run("newer stops startup and releases the lock", func(t *testing.T) { | |||
| 208 | + | dir := t.TempDir() | |||
| 209 | + | putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 2})) | |||
| 210 | + | _, err := kvstore.Open(ctx, dir) | |||
| 211 | + | require.ErrorIs(t, err, kvstore.ErrSchemaTooNew) | |||
| 212 | + | assert.Equal(t, uint32(2), readSchema(t, dir), "the lock was released and the value kept") | |||
| 213 | + | }) | |||
| 214 | + | ||||
| 215 | + | t.Run("garbage and zero are errors", func(t *testing.T) { | |||
| 216 | + | dir := t.TempDir() | |||
| 217 | + | putRaw(t, dir, "meta/schema", []byte{0xff, 0xff, 0xff}) | |||
| 218 | + | _, err := kvstore.Open(ctx, dir) | |||
| 219 | + | assert.Error(t, err) | |||
| 220 | + | ||||
| 221 | + | dir = t.TempDir() | |||
| 222 | + | putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 0})) | |||
| 223 | + | _, err = kvstore.Open(ctx, dir) | |||
| 224 | + | assert.Error(t, err) | |||
| 225 | + | }) | |||
| 226 | + | } | |||
| 227 | + | ||||
| 228 | + | func mustProto(t *testing.T, m proto.Message) []byte { | |||
| 229 | + | t.Helper() | |||
| 230 | + | b, err := proto.Marshal(m) | |||
| 231 | + | require.NoError(t, err) | |||
| 232 | + | return b | |||
| 233 | + | } | |||
| 234 | + | ||||
| 235 | + | func TestUnspecifiedVisibilityReadsPrivate(t *testing.T) { | |||
| 236 | + | ctx := context.Background() | |||
| 237 | + | dir := t.TempDir() | |||
| 238 | + | putRaw(t, dir, "repo/odd", mustProto(t, &storagepb.Repo{Name: "odd"})) // visibility unset | |||
| 239 | + | putRaw(t, dir, "repo/future", mustProto(t, &storagepb.Repo{Name: "future", Visibility: storagepb.Visibility(42)})) | |||
| 240 | + | s := openStore(t, dir) | |||
| 241 | + | ||||
| 242 | + | for _, n := range []string{"odd", "future"} { | |||
| 243 | + | got, err := s.Get(ctx, name(t, n)) | |||
| 244 | + | require.NoError(t, err) | |||
| 245 | + | assert.Equal(t, domain.VisibilityPrivate, got.Visibility, n) | |||
| 246 | + | } | |||
| 247 | + | } | |||
| 248 | + | ||||
| 249 | + | func TestCorruptRepoValue(t *testing.T) { | |||
| 250 | + | ctx := context.Background() | |||
| 251 | + | dir := t.TempDir() | |||
| 252 | + | putRaw(t, dir, "repo/bad", []byte{0xff, 0xff}) | |||
| 253 | + | putRaw(t, dir, "repo/mismatch", mustProto(t, &storagepb.Repo{Name: "Not Valid"})) | |||
| 254 | + | s := openStore(t, dir) | |||
| 255 | + | ||||
| 256 | + | _, err := s.Get(ctx, name(t, "bad")) | |||
| 257 | + | assert.Error(t, err) | |||
| 258 | + | _, err = s.Get(ctx, name(t, "mismatch")) | |||
| 259 | + | assert.ErrorIs(t, err, domain.ErrInvalidName) | |||
| 260 | + | _, err = s.List(ctx) | |||
| 261 | + | assert.Error(t, err) | |||
| 262 | + | } | |||