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.
5package kvstore
6
7import (
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. Version 2
21// adds branch prefixes to tokens: a build that does not know them would treat
22// a branch-limited token as unrestricted, so it must refuse the data
23// (ErrSchemaTooNew) instead.
24const SchemaVersion = 2
25
26const schemaKey = "meta/schema"
27
28// ErrSchemaTooNew means the data was written by a newer ermite.
29var ErrSchemaTooNew = errors.New("storage schema is newer than this ermite supports")
30
31// Store is the ermite-kv backed metadata store.
32type Store struct {
33 db *kv.DB
34}
35
36var _ ports.RepoMetaStore = (*Store)(nil)
37
38// Open opens (or creates) the database in dir and checks the storage schema:
39// a missing version is written as SchemaVersion, a lower one is upgraded to it,
40// a higher one stops with ErrSchemaTooNew. dir must exist.
41func Open(ctx context.Context, dir string) (*Store, error) {
42 db, err := kv.Open(ctx, dir, kv.Options{})
43 if err != nil {
44 return nil, fmt.Errorf("open ermite-kv in %s: %w", dir, err)
45 }
46 s := &Store{db: db}
47 if err := s.checkSchema(ctx); err != nil {
48 _ = db.Close(context.WithoutCancel(ctx))
49 return nil, err
50 }
51 return s, nil
52}
53
54// Close stops the database after the writes in progress.
55func (s *Store) Close(ctx context.Context) error {
56 if err := s.db.Close(ctx); err != nil {
57 return fmt.Errorf("close ermite-kv: %w", err)
58 }
59 return nil
60}
61
62// Check reports whether the database accepts writes. It returns the failure
63// once ermite-kv is closed or failed, and writes nothing.
64//
65// ermite-kv has no state accessor, and an empty batch sent through Apply would
66// queue behind the writes in progress. Apply looks at the state of the database
67// before it looks at its context, though, and an ended context stops it before
68// the writer channel: Apply on a cancelled context therefore answers ErrClosed
69// or ErrFailed when the database is stopped, and context.Canceled when it is
70// healthy, without waiting for the writer.
71func (s *Store) Check(ctx context.Context) error {
72 if err := ctx.Err(); err != nil {
73 return err
74 }
75 probe, cancel := context.WithCancel(context.WithoutCancel(ctx))
76 cancel()
77 var b kv.Batch
78 if _, err := s.db.Apply(probe, &b); err != nil && !errors.Is(err, context.Canceled) {
79 return fmt.Errorf("ermite-kv: %w", err)
80 }
81 return nil
82}
83
84func (s *Store) checkSchema(ctx context.Context) error {
85 log := zerowrap.FromCtx(ctx)
86 key := []byte(schemaKey)
87 for range 2 {
88 item, ok := s.db.Get(ctx, key)
89 if !ok {
90 var b kv.Batch
91 b.RequireAbsent(key)
92 b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: SchemaVersion}))
93 _, err := s.db.Apply(ctx, &b)
94 if errors.Is(err, kv.ErrConflict) {
95 continue // written meanwhile: read it
96 }
97 if err != nil {
98 return fmt.Errorf("write storage schema: %w", err)
99 }
100 log.Info().Uint32("schema", SchemaVersion).Msg("storage schema initialized")
101 return nil
102 }
103 var sv storagepb.SchemaVersion
104 if err := proto.Unmarshal(item.Value, &sv); err != nil {
105 return fmt.Errorf("decode storage schema: %w", err)
106 }
107 switch {
108 case sv.GetVersion() > SchemaVersion:
109 return fmt.Errorf("%w: found %d, supported %d", ErrSchemaTooNew, sv.GetVersion(), SchemaVersion)
110 case sv.GetVersion() == 0:
111 return fmt.Errorf("decode storage schema: invalid version 0")
112 }
113 if sv.GetVersion() < SchemaVersion {
114 // Older data stays readable as it is (the changes are additive);
115 // the new version is recorded so that older builds refuse it.
116 var b kv.Batch
117 b.RequireVersion(key, item.Version)
118 b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: SchemaVersion}))
119 _, err := s.db.Apply(ctx, &b)
120 if errors.Is(err, kv.ErrConflict) {
121 continue // changed meanwhile: read it again
122 }
123 if err != nil {
124 return fmt.Errorf("upgrade storage schema: %w", err)
125 }
126 log.Info().Uint32("from", sv.GetVersion()).Uint32("schema", SchemaVersion).Msg("storage schema upgraded")
127 return nil
128 }
129 log.Info().Uint32("schema", sv.GetVersion()).Msg("storage schema checked")
130 return nil
131 }
132 return errors.New("write storage schema: concurrent initialization did not settle")
133}
134
135func mustMarshal(m proto.Message) []byte {
136 b, err := proto.MarshalOptions{Deterministic: true}.Marshal(m)
137 if err != nil {
138 panic(fmt.Sprintf("kvstore: marshal %T: %v", m, err)) // only for invalid UTF-8 strings, checked by callers
139 }
140 return b
141}