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,
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 }
127 copy(a.Assignment[:], pb.GetAssignment())
128 for _, s := range artifactStates {
129 if s.p == pb.GetState() {
130 a.State = s.d
131 }
132 }
133 if a.State == "" {
134 return domain.CIArtifact{}, fmt.Errorf("decode artifact %d: unknown state %v", a.ID, pb.GetState())
135 }
136 return a, nil
137}
138
139// --- reads ---
140
141// artifactIDTries bounds the attempts of NextArtifactID: each conflict means
142// another call took an id, and uploads begin many at once.
143const artifactIDTries = 1000
144
145// NextArtifactID takes the next id in its own batch.
146func (c *CIArtifacts) NextArtifactID(ctx context.Context) (int64, error) {
147 ci := &CI{db: c.db}
148 for range artifactIDTries {
149 last, item, ok, err := ci.readSeq(ctx, []byte(artifactSeqKey))
150 if err != nil {
151 return 0, err
152 }
153 var b kv.Batch
154 stageSeq(&b, []byte(artifactSeqKey), item, ok, last+1)
155 _, err = c.db.Apply(ctx, &b)
156 if errors.Is(err, kv.ErrConflict) {
157 continue
158 }
159 if err != nil {
160 return 0, fmt.Errorf("take an artifact id: %w", err)
161 }
162 return last + 1, nil
163 }
164 return 0, fmt.Errorf("take an artifact id: %w", domain.ErrConflict)
165}
166
167// ArtifactUsage reads artifactuse/<repo>.
168func (c *CIArtifacts) ArtifactUsage(ctx context.Context, repo domain.RepoName) (int64, uint64, error) {
169 item, ok := c.db.Get(ctx, artifactUseKey(repo))
170 if !ok {
171 return 0, 0, nil
172 }
173 var pb storagepb.CIArtifactUsage
174 if err := proto.Unmarshal(item.Value, &pb); err != nil {
175 return 0, 0, fmt.Errorf("decode artifact usage: %w", err)
176 }
177 return pb.GetUsed(), item.Version, nil
178}
179
180// GetArtifact resolves the id index, then reads the record.
181func (c *CIArtifacts) GetArtifact(ctx context.Context, id int64) (domain.CIArtifact, error) {
182 snap := c.db.Snapshot(ctx)
183 idx, ok := snap.Get(ctx, artifactIDKey(id))
184 if !ok || id < 1 {
185 return domain.CIArtifact{}, domain.ErrNotFound
186 }
187 repo, runID, err := decodeArtifactIndex(idx.Value)
188 if err != nil {
189 return domain.CIArtifact{}, err
190 }
191 item, ok := snap.Get(ctx, artifactKey(repo, runID, id))
192 if !ok {
193 return domain.CIArtifact{}, domain.ErrNotFound
194 }
195 return decodeArtifact(item)
196}
197
198func encodeArtifactIndex(a domain.CIArtifact) []byte {
199 return []byte(a.Repo.String() + "/" + fmt.Sprintf(idFormat, a.RunID))
200}
201
202func decodeArtifactIndex(val []byte) (domain.RepoName, int64, error) {
203 name, run, ok := strings.Cut(string(val), "/")
204 if !ok {
205 return domain.RepoName{}, 0, fmt.Errorf("decode artifact index %q", val)
206 }
207 repo, err := domain.ParseRepoName(name)
208 if err != nil {
209 return domain.RepoName{}, 0, fmt.Errorf("decode artifact index: %w", err)
210 }
211 runID, err := strconv.ParseInt(run, 10, 64)
212 if err != nil {
213 return domain.RepoName{}, 0, fmt.Errorf("decode artifact index: %w", err)
214 }
215 return repo, runID, nil
216}
217
218// RunArtifacts lists artifact/<repo>/<run>/, which is in id order.
219func (c *CIArtifacts) RunArtifacts(ctx context.Context, repo domain.RepoName, runID int64) ([]domain.CIArtifact, error) {
220 return c.list(ctx, artifactRunPrefix(repo, runID))
221}
222
223// AllArtifacts lists artifact/.
224func (c *CIArtifacts) AllArtifacts(ctx context.Context) ([]domain.CIArtifact, error) {
225 return c.list(ctx, artifactPrefix)
226}
227
228func (c *CIArtifacts) list(ctx context.Context, prefix string) ([]domain.CIArtifact, error) {
229 var out []domain.CIArtifact
230 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
231 a, err := decodeArtifact(item)
232 if err != nil {
233 return nil, fmt.Errorf("decode %s: %w", key, err)
234 }
235 out = append(out, a)
236 }
237 return out, nil
238}
239
240// ArtifactBlocks lists artifactblk/<id>/.
241func (c *CIArtifacts) ArtifactBlocks(ctx context.Context, id int64) ([]ports.CIArtifactBlock, error) {
242 prefix := artifactBlocksPrefix(id)
243 var out []ports.CIArtifactBlock
244 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
245 block, err := hex.DecodeString(strings.TrimPrefix(string(key), prefix))
246 if err != nil {
247 return nil, fmt.Errorf("decode %s: %w", key, err)
248 }
249 var pb storagepb.CIArtifactBlock
250 if err := proto.Unmarshal(item.Value, &pb); err != nil {
251 return nil, fmt.Errorf("decode %s: %w", key, err)
252 }
253 out = append(out, ports.CIArtifactBlock{ArtifactID: id, ID: block, Size: pb.GetSize(), File: pb.GetFile()})
254 }
255 return out, nil
256}
257
258// BlockArtifactIDs lists the artifact ids under artifactblk/.
259func (c *CIArtifacts) BlockArtifactIDs(ctx context.Context) ([]int64, error) {
260 var ids []int64
261 for key := range c.db.Snapshot(ctx).Prefix(ctx, []byte(artifactBlkPrefix)) {
262 raw, _, _ := strings.Cut(strings.TrimPrefix(string(key), artifactBlkPrefix), "/")
263 id, err := strconv.ParseInt(raw, 10, 64)
264 if err != nil {
265 return nil, fmt.Errorf("decode %s: %w", key, err)
266 }
267 if len(ids) == 0 || ids[len(ids)-1] != id {
268 ids = append(ids, id)
269 }
270 }
271 return ids, nil
272}
273
274// --- writes ---
275
276// ApplyArtifacts writes w in one batch.
277func (c *CIArtifacts) ApplyArtifacts(ctx context.Context, w ports.CIArtifactWrite) error {
278 var b kv.Batch
279 guard := false
280 if w.UsageVersion == 0 {
281 b.RequireAbsent(artifactUseKey(w.Repo))
282 guard = true
283 } else {
284 b.RequireVersion(artifactUseKey(w.Repo), w.UsageVersion)
285 }
286 b.Put(artifactUseKey(w.Repo), mustMarshal(&storagepb.CIArtifactUsage{Used: w.Used}))
287 snap := c.db.Snapshot(ctx)
288 for _, a := range w.Put {
289 if a.Repo != w.Repo {
290 return fmt.Errorf("%w: artifact %d of another repository", domain.ErrInvalid, a.ID)
291 }
292 val, err := encodeArtifact(a)
293 if err != nil {
294 return err
295 }
296 stage(&b, artifactKey(a.Repo, a.RunID, a.ID), val, a.Version)
297 if a.Version == 0 {
298 b.RequireAbsent(artifactIDKey(a.ID))
299 b.Put(artifactIDKey(a.ID), encodeArtifactIndex(a))
300 guard = true
301 }
302 }
303 for _, a := range w.Delete {
304 if a.Repo != w.Repo || a.Version == 0 {
305 return fmt.Errorf("%w: delete of artifact %d", domain.ErrInvalid, a.ID)
306 }
307 key := artifactKey(a.Repo, a.RunID, a.ID)
308 b.RequireVersion(key, a.Version)
309 b.Delete(key)
310 b.Delete(artifactIDKey(a.ID))
311 for k := range snap.Prefix(ctx, []byte(artifactBlocksPrefix(a.ID))) {
312 b.Delete(slices.Clone(k))
313 }
314 }
315 for _, id := range w.DropBlocks {
316 for k := range snap.Prefix(ctx, []byte(artifactBlocksPrefix(id))) {
317 b.Delete(slices.Clone(k))
318 }
319 }
320 for _, blk := range w.PutBlocks {
321 if !utf8.ValidString(blk.File) {
322 return fmt.Errorf("%w: block file name", domain.ErrInvalid)
323 }
324 b.Put(artifactBlockKey(blk.ArtifactID, blk.ID), mustMarshal(&storagepb.CIArtifactBlock{Size: blk.Size, File: blk.File}))
325 }
326 if guard {
327 if err := guardRepoKey(ctx, c.db, &b, w.Repo); err != nil {
328 return err
329 }
330 }
331 _, err := c.db.Apply(ctx, &b)
332 if errors.Is(err, kv.ErrConflict) {
333 return domain.ErrConflict
334 }
335 if err != nil {
336 return fmt.Errorf("write the artifacts of %q: %w", w.Repo, err)
337 }
338 return nil
339}
340
341// artifactKeys lists every key of the snapshot that belongs to the artifacts
342// of the repository: records, their id index and blocks, and the usage record.
343func artifactKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
344 keys := [][]byte{artifactUseKey(repo)}
345 for key, item := range snap.Prefix(ctx, []byte(artifactPrefix+repo.String()+"/")) {
346 keys = append(keys, slices.Clone(key))
347 if a, err := decodeArtifact(item); err == nil {
348 keys = append(keys, artifactIDKey(a.ID))
349 for k := range snap.Prefix(ctx, []byte(artifactBlocksPrefix(a.ID))) {
350 keys = append(keys, slices.Clone(k))
351 }
352 }
353 }
354 return keys
355}
356
357// DeleteBlocks removes artifactblk/<id>/.
358func (c *CIArtifacts) DeleteBlocks(ctx context.Context, id int64) error {
359 var b kv.Batch
360 for k := range c.db.Snapshot(ctx).Prefix(ctx, []byte(artifactBlocksPrefix(id))) {
361 b.Delete(slices.Clone(k))
362 }
363 if b.Len() == 0 {
364 return nil
365 }
366 if _, err := c.db.Apply(ctx, &b); err != nil {
367 return fmt.Errorf("delete the blocks of artifact %d: %w", id, err)
368 }
369 return nil
370}