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