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 b := pendingArtifact(n, 7, id2, "logs")
44 require.NoError(t, arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Put: []domain.CIArtifact{a, b}}))
45
46 got, err := arts.GetArtifact(ctx, id1)
47 require.NoError(t, err)
48 assert.Equal(t, "dist", got.Name)
49 assert.Equal(t, a.Assignment, got.Assignment)
50 assert.Equal(t, domain.CIArtifactPending, got.State)
51 assert.True(t, got.ExpiresAt.Equal(a.ExpiresAt))
52 assert.NotZero(t, got.Version)
53
54 list, err := arts.RunArtifacts(ctx, n, 7)
55 require.NoError(t, err)
56 require.Len(t, list, 2)
57 assert.Equal(t, []int64{id1, id2}, []int64{list[0].ID, list[1].ID}, "by id")
58 other, err := arts.RunArtifacts(ctx, n, 8)
59 require.NoError(t, err)
60 assert.Empty(t, other)
61
62 // A second create of the same id is a conflict.
63 err = arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, UsageVersion: 0, Put: []domain.CIArtifact{a}})
64 assert.ErrorIs(t, err, domain.ErrConflict)
65
66 // Blocks, and the usage precondition.
67 _, ver, err = arts.ArtifactUsage(ctx, n)
68 require.NoError(t, err)
69 got.Charged = 10
70 blk := ports.CIArtifactBlock{ArtifactID: id1, ID: []byte{0, 1, 0xff}, Size: 10, File: "0001ff.ab"}
71 require.NoError(t, arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 10, UsageVersion: ver, Put: []domain.CIArtifact{got}, PutBlocks: []ports.CIArtifactBlock{blk}}))
72 err = arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 99, UsageVersion: ver})
73 assert.ErrorIs(t, err, domain.ErrConflict, "the usage moved since it was read")
74 used, _, err = arts.ArtifactUsage(ctx, n)
75 require.NoError(t, err)
76 assert.EqualValues(t, 10, used)
77 blocks, err := arts.ArtifactBlocks(ctx, id1)
78 require.NoError(t, err)
79 assert.Equal(t, []ports.CIArtifactBlock{blk}, blocks)
80 ids, err := arts.BlockArtifactIDs(ctx)
81 require.NoError(t, err)
82 assert.Equal(t, []int64{id1}, ids)
83
84 // A stale record version is a conflict too.
85 _, ver, _ = arts.ArtifactUsage(ctx, n)
86 err = arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 10, UsageVersion: ver, Put: []domain.CIArtifact{got}})
87 assert.ErrorIs(t, err, domain.ErrConflict)
88
89 // Delete takes the record, its index and its blocks.
90 cur, err := arts.GetArtifact(ctx, id1)
91 require.NoError(t, err)
92 _, ver, _ = arts.ArtifactUsage(ctx, n)
93 require.NoError(t, arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 0, UsageVersion: ver, Delete: []domain.CIArtifact{cur}}))
94 _, err = arts.GetArtifact(ctx, id1)
95 assert.ErrorIs(t, err, domain.ErrNotFound)
96 blocks, err = arts.ArtifactBlocks(ctx, id1)
97 require.NoError(t, err)
98 assert.Empty(t, blocks)
99 all, err := arts.AllArtifacts(ctx)
100 require.NoError(t, err)
101 require.Len(t, all, 1)
102 assert.Equal(t, id2, all[0].ID)
103
104 _, err = arts.GetArtifact(ctx, 0)
105 assert.ErrorIs(t, err, domain.ErrNotFound)
106}
107
108func TestCIArtifactIDsAreUniqueUnderConcurrency(t *testing.T) {
109 ctx := context.Background()
110 s, _, _ := ciRepo(t)
111 arts := s.Artifacts()
112 var mu sync.Mutex
113 seen := map[int64]bool{}
114 var wg sync.WaitGroup
115 for range 8 {
116 wg.Go(func() {
117 for range 10 {
118 id, err := arts.NextArtifactID(ctx)
119 assert.NoError(t, err)
120 mu.Lock()
121 assert.False(t, seen[id], "id %d given twice", id)
122 seen[id] = true
123 mu.Unlock()
124 }
125 })
126 }
127 wg.Wait()
128 assert.Len(t, seen, 80)
129}
130
131func TestCIArtifactsGoWithTheRepository(t *testing.T) {
132 ctx := context.Background()
133 s, _, n := ciRepo(t)
134 arts := s.Artifacts()
135 a := pendingArtifact(n, 3, 1, "dist")
136 blk := ports.CIArtifactBlock{ArtifactID: 1, ID: []byte("b"), Size: 4, File: "62.00"}
137 require.NoError(t, arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 4, Put: []domain.CIArtifact{a}, PutBlocks: []ports.CIArtifactBlock{blk}}))
138
139 require.NoError(t, s.Delete(ctx, n))
140 _, err := arts.GetArtifact(ctx, 1)
141 assert.ErrorIs(t, err, domain.ErrNotFound)
142 all, err := arts.AllArtifacts(ctx)
143 require.NoError(t, err)
144 assert.Empty(t, all)
145 ids, err := arts.BlockArtifactIDs(ctx)
146 require.NoError(t, err)
147 assert.Empty(t, ids)
148 used, ver, err := arts.ArtifactUsage(ctx, n)
149 require.NoError(t, err)
150 assert.Zero(t, used)
151 assert.Zero(t, ver)
152
153 // No write recreates keys under the deleted repository.
154 err = arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Put: []domain.CIArtifact{pendingArtifact(n, 3, 2, "x")}})
155 assert.ErrorIs(t, err, domain.ErrNotFound)
156 all, _ = arts.AllArtifacts(ctx)
157 assert.Empty(t, all)
158}
159
160func TestCIArtifactDeleteBlocks(t *testing.T) {
161 ctx := context.Background()
162 s, _, n := ciRepo(t)
163 arts := s.Artifacts()
164 a := pendingArtifact(n, 3, 5, "dist")
165 blks := []ports.CIArtifactBlock{{ArtifactID: 5, ID: []byte("a"), Size: 1, File: "61.00"}, {ArtifactID: 5, ID: []byte("b"), Size: 1, File: "62.00"}}
166 require.NoError(t, arts.ApplyArtifacts(ctx, ports.CIArtifactWrite{Repo: n, Used: 2, Put: []domain.CIArtifact{a}, PutBlocks: blks}))
167 require.NoError(t, arts.DeleteBlocks(ctx, 5))
168 got, err := arts.ArtifactBlocks(ctx, 5)
169 require.NoError(t, err)
170 assert.Empty(t, got)
171 require.NoError(t, arts.DeleteBlocks(ctx, 5), "nothing left is fine")
172}