1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore_test
4
5import (
6 "context"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/assert"
12 "github.com/stretchr/testify/require"
13
14 "bnema.dev/ermite/internal/domain"
15 "bnema.dev/ermite/internal/ports"
16)
17
18func pendingArtifact(repo domain.RepoName, run, id int64, name string) domain.CIArtifact {
19 return domain.CIArtifact{
20 ID: id, Repo: repo, RunID: run, Attempt: 1, JobID: "1-0-0", Assignment: domain.HashTokenSecret("ermitejob_x"),
21 Name: name, MediaType: domain.CIArtifactZip, State: domain.CIArtifactPending,
22 CreatedAt: created, ExpiresAt: created.Add(24 * time.Hour),
23 }
24}
25
26func TestCIArtifactsRoundTripAndIndexes(t *testing.T) {
27 ctx := context.Background()
28 s, _, n := ciRepo(t)
29 arts := s.Artifacts()
30
31 used, ver, err := arts.ArtifactUsage(ctx, n)
32 require.NoError(t, err)
33 assert.Zero(t, used)
34 assert.Zero(t, ver)
35
36 id1, err := arts.NextArtifactID(ctx)
37 require.NoError(t, err)
38 id2, err := arts.NextArtifactID(ctx)
39 require.NoError(t, err)
40 assert.Greater(t, id2, id1, "ids increase")
41
42 a := pendingArtifact(n, 7, id1, "dist")
43 a.Store = domain.CIArtifactStoreBlob
44 b := pendingArtifact(n, 7, id2, "logs")
45 require.NoError(t, arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Put: []domain.CIArtifact{a, b}}))
46
47 got, err := arts.GetArtifact(ctx, id1)
48 require.NoError(t, err)
49 assert.Equal(t, "dist", got.Name)
50 assert.Equal(t, domain.CIArtifactStoreBlob, got.Store)
51 legacy, err := arts.GetArtifact(ctx, id2)
52 require.NoError(t, err)
53 assert.Equal(t, domain.CIArtifactStoreLegacy, legacy.Store, "an absent proto field keeps the legacy layout")
54 assert.Equal(t, a.Assignment, got.Assignment)
55 assert.Equal(t, domain.CIArtifactPending, got.State)
56 assert.True(t, got.ExpiresAt.Equal(a.ExpiresAt))
57 assert.NotZero(t, got.Version)
58
59 list, err := arts.RunArtifacts(ctx, n, 7)
60 require.NoError(t, err)
61 require.Len(t, list, 2)
62 assert.Equal(t, []int64{id1, id2}, []int64{list[0].ID, list[1].ID}, "by id")
63 other, err := arts.RunArtifacts(ctx, n, 8)
64 require.NoError(t, err)
65 assert.Empty(t, other)
66
67 // A second create of the same id is a conflict.
68 err = arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, UsageVersion: 0, Put: []domain.CIArtifact{a}})
69 assert.ErrorIs(t, err, domain.ErrConflict)
70
71 // Blocks, and the usage precondition.
72 _, ver, err = arts.ArtifactUsage(ctx, n)
73 require.NoError(t, err)
74 got.Charged = 10
75 blk := ports.CIArtifactBlock{ArtifactID: id1, ID: []byte{0, 1, 0xff}, Size: 10, File: "0001ff.ab"}
76 require.NoError(t, arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 10, UsageVersion: ver, Put: []domain.CIArtifact{got}, PutBlocks: []ports.CIArtifactBlock{blk}}))
77 err = arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 99, UsageVersion: ver})
78 assert.ErrorIs(t, err, domain.ErrConflict, "the usage moved since it was read")
79 used, _, err = arts.ArtifactUsage(ctx, n)
80 require.NoError(t, err)
81 assert.EqualValues(t, 10, used)
82 blocks, err := arts.ArtifactBlocks(ctx, id1)
83 require.NoError(t, err)
84 assert.Equal(t, []ports.CIArtifactBlock{blk}, blocks)
85 ids, err := arts.BlockArtifactIDs(ctx)
86 require.NoError(t, err)
87 assert.Equal(t, []int64{id1}, ids)
88
89 // A stale record version is a conflict too.
90 _, ver, _ = arts.ArtifactUsage(ctx, n)
91 err = arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 10, UsageVersion: ver, Put: []domain.CIArtifact{got}})
92 assert.ErrorIs(t, err, domain.ErrConflict)
93
94 // Delete takes the record, its index and its blocks.
95 cur, err := arts.GetArtifact(ctx, id1)
96 require.NoError(t, err)
97 _, ver, _ = arts.ArtifactUsage(ctx, n)
98 require.NoError(t, arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 0, UsageVersion: ver, Delete: []domain.CIArtifact{cur}}))
99 _, err = arts.GetArtifact(ctx, id1)
100 assert.ErrorIs(t, err, domain.ErrNotFound)
101 blocks, err = arts.ArtifactBlocks(ctx, id1)
102 require.NoError(t, err)
103 assert.Empty(t, blocks)
104 all, err := arts.AllArtifacts(ctx)
105 require.NoError(t, err)
106 require.Len(t, all, 1)
107 assert.Equal(t, id2, all[0].ID)
108
109 _, err = arts.GetArtifact(ctx, 0)
110 assert.ErrorIs(t, err, domain.ErrNotFound)
111}
112
113func TestCIArtifactIDsAreUniqueUnderConcurrency(t *testing.T) {
114 ctx := context.Background()
115 s, _, _ := ciRepo(t)
116 arts := s.Artifacts()
117 var mu sync.Mutex
118 seen := map[int64]bool{}
119 var wg sync.WaitGroup
120 for range 8 {
121 wg.Go(func() {
122 for range 10 {
123 id, err := arts.NextArtifactID(ctx)
124 assert.NoError(t, err)
125 mu.Lock()
126 assert.False(t, seen[id], "id %d given twice", id)
127 seen[id] = true
128 mu.Unlock()
129 }
130 })
131 }
132 wg.Wait()
133 assert.Len(t, seen, 80)
134}
135
136func TestCIArtifactsGoWithTheRepository(t *testing.T) {
137 ctx := context.Background()
138 s, _, n := ciRepo(t)
139 arts := s.Artifacts()
140 a := pendingArtifact(n, 3, 1, "dist")
141 blk := ports.CIArtifactBlock{ArtifactID: 1, ID: []byte("b"), Size: 4, File: "62.00"}
142 require.NoError(t, arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 4, Put: []domain.CIArtifact{a}, PutBlocks: []ports.CIArtifactBlock{blk}}))
143
144 require.NoError(t, s.Delete(ctx, n))
145 _, err := arts.GetArtifact(ctx, 1)
146 assert.ErrorIs(t, err, domain.ErrNotFound)
147 all, err := arts.AllArtifacts(ctx)
148 require.NoError(t, err)
149 assert.Empty(t, all)
150 ids, err := arts.BlockArtifactIDs(ctx)
151 require.NoError(t, err)
152 assert.Empty(t, ids)
153 used, ver, err := arts.ArtifactUsage(ctx, n)
154 require.NoError(t, err)
155 assert.Zero(t, used)
156 assert.Zero(t, ver)
157
158 // No write recreates keys under the deleted repository.
159 err = arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Put: []domain.CIArtifact{pendingArtifact(n, 3, 2, "x")}})
160 assert.ErrorIs(t, err, domain.ErrNotFound)
161 all, _ = arts.AllArtifacts(ctx)
162 assert.Empty(t, all)
163}
164
165func TestCIArtifactDeleteBlocks(t *testing.T) {
166 ctx := context.Background()
167 s, _, n := ciRepo(t)
168 arts := s.Artifacts()
169 a := pendingArtifact(n, 3, 5, "dist")
170 blks := []ports.CIArtifactBlock{{ArtifactID: 5, ID: []byte("a"), Size: 1, File: "61.00"}, {ArtifactID: 5, ID: []byte("b"), Size: 1, File: "62.00"}}
171 require.NoError(t, arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 2, Put: []domain.CIArtifact{a}, PutBlocks: blks}))
172 require.NoError(t, arts.DeleteBlocks(ctx, 5))
173 got, err := arts.ArtifactBlocks(ctx, 5)
174 require.NoError(t, err)
175 assert.Empty(t, got)
176 require.NoError(t, arts.DeleteBlocks(ctx, 5), "nothing left is fine")
177}