1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "encoding/hex"
8 "errors"
9 "fmt"
10 "slices"
11 "strconv"
12 "strings"
13 "unicode/utf8"
14
15 kv "bnema.dev/ermite-kv"
16 "google.golang.org/protobuf/proto"
17
18 storagepb "bnema.dev/ermite/gen/ermite/storage"
19 "bnema.dev/ermite/internal/domain"
20 "bnema.dev/ermite/internal/ports"
21)
22
23// Key layout of the CI artifacts (CI-4 D3). New prefixes only: an older build
24// ignores them, so there is no schema change.
25//
26// artifact/<repo>/<run:020d>/<id:020d> CIArtifact
27// artifactid/<id:020d> index: "<repo>/<run:020d>"
28// artifactblk/<id:020d>/<hex(block id)> CIArtifactBlock: a staged block
29// artifactuse/<repo> CIArtifactUsage: bytes charged
30// artifactseq Sequence: last artifact id
31//
32// Every write of the artifacts of a repository requires and rewrites its
33// usage record (ports.CIArtifactWrite). The records of a repository go with it
34// (ciKeys); those of a run that retention removed are deleted by the sweep of
35// the artifacts, which refunds them.
36const (
37 artifactPrefix = "artifact/"
38 artifactIDPrefix = "artifactid/"
39 artifactBlkPrefix = "artifactblk/"
40 artifactUsePrefix = "artifactuse/"
41 artifactSeqKey = "artifactseq"
42)
43
44func artifactRunPrefix(repo domain.RepoName, runID int64) string {
45 return artifactPrefix + repo.String() + "/" + fmt.Sprintf(idFormat, runID) + "/"
46}
47func artifactKey(repo domain.RepoName, runID, id int64) []byte {
48 return []byte(artifactRunPrefix(repo, runID) + fmt.Sprintf(idFormat, id))
49}
50func artifactIDKey(id int64) []byte { return []byte(artifactIDPrefix + fmt.Sprintf(idFormat, id)) }
51func artifactBlocksPrefix(id int64) string {
52 return artifactBlkPrefix + fmt.Sprintf(idFormat, id) + "/"
53}
54func artifactBlockKey(id int64, block []byte) []byte {
55 return []byte(artifactBlocksPrefix(id) + hex.EncodeToString(block))
56}
57func artifactUseKey(repo domain.RepoName) []byte { return []byte(artifactUsePrefix + repo.String()) }
58
59// CIArtifacts is the ermite-kv backed ports.CIArtifactStore. It shares the
60// database of the Store it comes from.
61type CIArtifacts struct {
62 db *kv.DB
63}
64
65var _ ports.CIArtifactStore = (*CIArtifacts)(nil)
66
67// Artifacts returns the artifact store over the same database. It is valid
68// until the Store is closed.
69func (s *Store) Artifacts() *CIArtifacts { return &CIArtifacts{db: s.db} }
70
71// --- encoding ---
72
73var artifactStates = []struct {
74 d domain.CIArtifactState
75 p storagepb.CIArtifactState
76}{
77 {domain.CIArtifactPending, storagepb.CIArtifactState_CI_ARTIFACT_STATE_PENDING},
78 {domain.CIArtifactCommitted, storagepb.CIArtifactState_CI_ARTIFACT_STATE_COMMITTED},
79 {domain.CIArtifactFinal, storagepb.CIArtifactState_CI_ARTIFACT_STATE_FINAL},
80}
81
82// encodeArtifact serializes an artifact for storage.
83//
84//hexcheck:ignore suspicious-business-logic-in-adapter validates artifact storage identifiers, UTF-8 and enum encoding before protobuf serialization
85func encodeArtifact(a domain.CIArtifact) ([]byte, error) {
86 if a.Repo.IsZero() || a.ID < 1 || a.RunID < 1 {
87 return nil, fmt.Errorf("%w: artifact without id, run or repository", domain.ErrInvalid)
88 }
89 if !utf8.ValidString(a.Name) || !utf8.ValidString(a.MediaType) || !utf8.ValidString(a.JobID) {
90 return nil, fmt.Errorf("%w: artifact strings must be valid UTF-8", domain.ErrInvalid)
91 }
92 pb := &storagepb.CIArtifact{
93 Id: a.ID, Repo: a.Repo.String(), RunId: a.RunID, Attempt: a.Attempt, JobId: a.JobID,
94 Name: a.Name, MediaType: a.MediaType, Size: a.Size, Digest: a.Digest,
95 CreatedAt: encodeTime(a.CreatedAt), ExpiresAt: encodeTime(a.ExpiresAt),
96 Charged: a.Charged, ListDigest: a.ListDigest, Store: string(a.Store),
97 }
98 if a.Assignment != (domain.TokenHash{}) {
99 pb.Assignment = a.Assignment[:]
100 }
101 for _, s := range artifactStates {
102 if s.d == a.State {
103 pb.State = s.p
104 }
105 }
106 if pb.State == storagepb.CIArtifactState_CI_ARTIFACT_STATE_UNSPECIFIED {
107 return nil, fmt.Errorf("%w: artifact state %q", domain.ErrInvalid, a.State)
108 }
109 return mustMarshal(pb), nil
110}
111
112func decodeArtifact(item kv.Item) (domain.CIArtifact, error) {
113 var pb storagepb.CIArtifact
114 if err := proto.Unmarshal(item.Value, &pb); err != nil {
115 return domain.CIArtifact{}, fmt.Errorf("decode artifact: %w", err)
116 }
117 repo, err := domain.ParseRepoName(pb.GetRepo())
118 if err != nil {
119 return domain.CIArtifact{}, fmt.Errorf("decode artifact: %w", err)
120 }
121 a := domain.CIArtifact{
122 ID: pb.GetId(), Repo: repo, RunID: pb.GetRunId(), Attempt: pb.GetAttempt(), JobID: pb.GetJobId(),
123 Name: pb.GetName(), MediaType: pb.GetMediaType(), Size: pb.GetSize(), Digest: pb.GetDigest(),
124 CreatedAt: decodeTime(pb.GetCreatedAt()), ExpiresAt: decodeTime(pb.GetExpiresAt()),
125 Charged: pb.GetCharged(), ListDigest: pb.GetListDigest(), Version: item.Version,
126 Store: domain.CIArtifactStore(pb.GetStore()),
127 }
128 copy(a.Assignment[:], pb.GetAssignment())
129 for _, s := range artifactStates {
130 if s.p == pb.GetState() {
131 a.State = s.d
132 }
133 }
134 if a.State == "" {
135 return domain.CIArtifact{}, fmt.Errorf("decode artifact %d: unknown state %v", a.ID, pb.GetState())
136 }
137 return a, nil
138}
139
140// --- reads ---
141
142// artifactIDTries bounds the attempts of NextArtifactID: each conflict means
143// another call took an id, and uploads begin many at once.
144const artifactIDTries = 1000
145
146// NextArtifactID takes the next id in its own batch.
147func (c *CIArtifacts) NextArtifactID(ctx context.Context) (int64, error) {
148 ci := &CI{db: c.db}
149 for range artifactIDTries {
150 last, item, ok, err := ci.readSeq(ctx, []byte(artifactSeqKey))
151 if err != nil {
152 return 0, err
153 }
154 var b kv.Batch
155 stageSeq(&b, []byte(artifactSeqKey), item, ok, last+1)
156 _, err = c.db.Apply(ctx, &b)
157 if errors.Is(err, kv.ErrConflict) {
158 continue
159 }
160 if err != nil {
161 return 0, fmt.Errorf("take an artifact id: %w", err)
162 }
163 return last + 1, nil
164 }
165 return 0, fmt.Errorf("take an artifact id: %w", domain.ErrConflict)
166}
167
168// ArtifactUsage reads artifactuse/<repo>.
169func (c *CIArtifacts) ArtifactUsage(ctx context.Context, repo domain.RepoName) (int64, uint64, error) {
170 item, ok := c.db.Get(ctx, artifactUseKey(repo))
171 if !ok {
172 return 0, 0, nil
173 }
174 var pb storagepb.CIArtifactUsage
175 if err := proto.Unmarshal(item.Value, &pb); err != nil {
176 return 0, 0, fmt.Errorf("decode artifact usage: %w", err)
177 }
178 return pb.GetUsed(), item.Version, nil
179}
180
181// GetArtifact resolves the id index, then reads the record.
182func (c *CIArtifacts) GetArtifact(ctx context.Context, id int64) (domain.CIArtifact, error) {
183 snap := c.db.Snapshot(ctx)
184 idx, ok := snap.Get(ctx, artifactIDKey(id))
185 if !ok || id < 1 {
186 return domain.CIArtifact{}, domain.ErrNotFound
187 }
188 repo, runID, err := decodeArtifactIndex(idx.Value)
189 if err != nil {
190 return domain.CIArtifact{}, err
191 }
192 item, ok := snap.Get(ctx, artifactKey(repo, runID, id))
193 if !ok {
194 return domain.CIArtifact{}, domain.ErrNotFound
195 }
196 return decodeArtifact(item)
197}
198
199func encodeArtifactIndex(a domain.CIArtifact) []byte {
200 return []byte(a.Repo.String() + "/" + fmt.Sprintf(idFormat, a.RunID))
201}
202
203func decodeArtifactIndex(val []byte) (domain.RepoName, int64, error) {
204 name, run, ok := strings.Cut(string(val), "/")
205 if !ok {
206 return domain.RepoName{}, 0, fmt.Errorf("decode artifact index %q", val)
207 }
208 repo, err := domain.ParseRepoName(name)
209 if err != nil {
210 return domain.RepoName{}, 0, fmt.Errorf("decode artifact index: %w", err)
211 }
212 runID, err := strconv.ParseInt(run, 10, 64)
213 if err != nil {
214 return domain.RepoName{}, 0, fmt.Errorf("decode artifact index: %w", err)
215 }
216 return repo, runID, nil
217}
218
219// RunArtifacts lists artifact/<repo>/<run>/, which is in id order.
220func (c *CIArtifacts) RunArtifacts(ctx context.Context, repo domain.RepoName, runID int64) ([]domain.CIArtifact, error) {
221 return c.list(ctx, artifactRunPrefix(repo, runID))
222}
223
224// AllArtifacts lists artifact/.
225func (c *CIArtifacts) AllArtifacts(ctx context.Context) ([]domain.CIArtifact, error) {
226 return c.list(ctx, artifactPrefix)
227}
228
229func (c *CIArtifacts) list(ctx context.Context, prefix string) ([]domain.CIArtifact, error) {
230 var out []domain.CIArtifact
231 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
232 a, err := decodeArtifact(item)
233 if err != nil {
234 return nil, fmt.Errorf("decode %s: %w", key, err)
235 }
236 out = append(out, a)
237 }
238 return out, nil
239}
240
241// ArtifactBlocks lists artifactblk/<id>/.
242func (c *CIArtifacts) ArtifactBlocks(ctx context.Context, id int64) ([]ports.CIArtifactBlock, error) {
243 prefix := artifactBlocksPrefix(id)
244 var out []ports.CIArtifactBlock
245 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
246 block, err := hex.DecodeString(strings.TrimPrefix(string(key), prefix))
247 if err != nil {
248 return nil, fmt.Errorf("decode %s: %w", key, err)
249 }
250 var pb storagepb.CIArtifactBlock
251 if err := proto.Unmarshal(item.Value, &pb); err != nil {
252 return nil, fmt.Errorf("decode %s: %w", key, err)
253 }
254 out = append(out, ports.CIArtifactBlock{ArtifactID: id, ID: block, Size: pb.GetSize(), File: pb.GetFile()})
255 }
256 return out, nil
257}
258
259// BlockArtifactIDs lists the artifact ids under artifactblk/.
260func (c *CIArtifacts) BlockArtifactIDs(ctx context.Context) ([]int64, error) {
261 var ids []int64
262 for key := range c.db.Snapshot(ctx).Prefix(ctx, []byte(artifactBlkPrefix)) {
263 raw, _, _ := strings.Cut(strings.TrimPrefix(string(key), artifactBlkPrefix), "/")
264 id, err := strconv.ParseInt(raw, 10, 64)
265 if err != nil {
266 return nil, fmt.Errorf("decode %s: %w", key, err)
267 }
268 if len(ids) == 0 || ids[len(ids)-1] != id {
269 ids = append(ids, id)
270 }
271 }
272 return ids, nil
273}
274
275// --- writes ---
276
277// ApplyArtifacts writes w in one batch.
278func (c *CIArtifacts) ApplyArtifacts(ctx context.Context, w ports.CIArtifactWrite) error {
279 var b kv.Batch
280 guard := false
281 if w.UsageVersion == 0 {
282 b.RequireAbsent(artifactUseKey(w.Repo))
283 guard = true
284 } else {
285 b.RequireVersion(artifactUseKey(w.Repo), w.UsageVersion)
286 }
287 b.Put(artifactUseKey(w.Repo), mustMarshal(&storagepb.CIArtifactUsage{Used: w.Used}))
288 snap := c.db.Snapshot(ctx)
289 for _, a := range w.Put {
290 if a.Repo != w.Repo {
291 return fmt.Errorf("%w: artifact %d of another repository", domain.ErrInvalid, a.ID)
292 }
293 val, err := encodeArtifact(a)
294 if err != nil {
295 return err
296 }
297 stage(&b, artifactKey(a.Repo, a.RunID, a.ID), val, a.Version)
298 if a.Version == 0 {
299 b.RequireAbsent(artifactIDKey(a.ID))
300 b.Put(artifactIDKey(a.ID), encodeArtifactIndex(a))
301 guard = true
302 }
303 }
304 for _, a := range w.Delete {
305 if a.Repo != w.Repo || a.Version == 0 {
306 return fmt.Errorf("%w: delete of artifact %d", domain.ErrInvalid, a.ID)
307 }
308 key := artifactKey(a.Repo, a.RunID, a.ID)
309 b.RequireVersion(key, a.Version)
310 b.Delete(key)
311 b.Delete(artifactIDKey(a.ID))
312 for k := range snap.Prefix(ctx, []byte(artifactBlocksPrefix(a.ID))) {
313 b.Delete(slices.Clone(k))
314 }
315 }
316 for _, id := range w.DropBlocks {
317 for k := range snap.Prefix(ctx, []byte(artifactBlocksPrefix(id))) {
318 b.Delete(slices.Clone(k))
319 }
320 }
321 for _, blk := range w.PutBlocks {
322 if !utf8.ValidString(blk.File) {
323 return fmt.Errorf("%w: block file name", domain.ErrInvalid)
324 }
325 b.Put(artifactBlockKey(blk.ArtifactID, blk.ID), mustMarshal(&storagepb.CIArtifactBlock{Size: blk.Size, File: blk.File}))
326 }
327 if guard {
328 if err := guardRepoKey(ctx, c.db, &b, w.Repo); err != nil {
329 return err
330 }
331 }
332 _, err := c.db.Apply(ctx, &b)
333 if errors.Is(err, kv.ErrConflict) {
334 return domain.ErrConflict
335 }
336 if err != nil {
337 return fmt.Errorf("write the artifacts of %q: %w", w.Repo, err)
338 }
339 return nil
340}
341
342// artifactKeys lists every key of the snapshot that belongs to the artifacts
343// of the repository: records, their id index and blocks, and the usage record.
344func artifactKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
345 keys := [][]byte{artifactUseKey(repo)}
346 for key, item := range snap.Prefix(ctx, []byte(artifactPrefix+repo.String()+"/")) {
347 keys = append(keys, slices.Clone(key))
348 if a, err := decodeArtifact(item); err == nil {
349 keys = append(keys, artifactIDKey(a.ID))
350 for k := range snap.Prefix(ctx, []byte(artifactBlocksPrefix(a.ID))) {
351 keys = append(keys, slices.Clone(k))
352 }
353 }
354 }
355 return keys
356}
357
358// DeleteBlocks removes artifactblk/<id>/.
359func (c *CIArtifacts) DeleteBlocks(ctx context.Context, id int64) error {
360 var b kv.Batch
361 for k := range c.db.Snapshot(ctx).Prefix(ctx, []byte(artifactBlocksPrefix(id))) {
362 b.Delete(slices.Clone(k))
363 }
364 if b.Len() == 0 {
365 return nil
366 }
367 if _, err := c.db.Apply(ctx, &b); err != nil {
368 return fmt.Errorf("delete the blocks of artifact %d: %w", id, err)
369 }
370 return nil
371}